sqlalchemy mixin：after_create 未在子进程中触发

Question

我正在开发 pq 库（PostgreSQL 驱动的 python 队列系统）的 ORM 风格版本，用户可以在其中有自己的队列模型。它还添加了批量插入/获取、异步支持等功能（如果一切顺利，我将能够发布它）。

我在创建触发器（我使用 PostgreSQL 通知系统）创建表后自动时遇到困难（我想让使用尽可能简单，这样会好得多而不是添加一个额外的类方法来创建触发器）。

这与此帖子中的答案类似，但是我无法使用此解决方案，因为我需要传递连接< /strong> （用于通过检查连接的方言来转义 SQL 标识符以及检查对象是否预先存在）。

这是我根据我之前提到的帖子所做的尝试。我为冗长的代码表示歉意，但我认为我必须包含一些上下文。

基本模型

from sqlalchemy import (BIGINT, Column, func, Index, nullslast,
                        nullsfirst, SMALLINT, TIMESTAMP)
from sqlalchemy.orm import declared_attr, declarative_mixin
from sqlalchemy.event import listens_for

# this is the function that returns the base model
def postgres_queue_base(schema:str='public', tz_aware:bool=True, use_trigger:bool=True) -> 'PostgresQueueBase':

    @declarative_mixin  # this is only for MyPy, it does not modify anything
    class PostgresQueueBase:

        __tablename__ = 'queue'

        @declared_attr
        def __table_args__(cls):
            return (Index(nullsfirst(cls.schedule_at), nullslast(cls.dequeued_at), postgresql_where=(cls.dequeued_at == None)),
                    {'schema':schema})

        id = Column('id', BIGINT, primary_key=True)
        internal_mapping = Column('internal_mapping', BIGINT, nullable=False)
        enqueued_at = Column('enqueued_at', TIMESTAMP(timezone=tz_aware), nullable=False, server_default=func.now())
        dequeued_at = Column('dequeued_at', TIMESTAMP(timezone=tz_aware))
        expected_at = Column(TIMESTAMP(timezone=tz_aware))
        schedule_at = Column(TIMESTAMP(timezone=tz_aware))
        status = Column(SMALLINT, index=True)

    @listens_for(PostgresQueueBase, "instrument_class", propagate=True)
    def instrument_class(mapper, class_):
        print('EVENT INSTRUMENT CLASS')
        if use_trigger and mapper.local_table is not None:
            trigger_for_table(table=mapper.local_table)

    def trigger_for_table(table):
        print('Registering after_create event')

        @listens_for(table, "after_create")
        def create_trigger(table, connection):
            print('AFTER CREATE EVENT')
            # code that creates triggers and logs that (here I'll just print something and put pseudo code in a comment)
            # trig = PostgresQueueTrigger(schema=get_schema_from_model(table), table_name=table.name, connection=connection)
            # trig.add_trigger()
            print('Creating notify function public.notify_job')
            # unique trigger name using hash of schema.table_name (avoids problems with long names and special chars)
            print('Creating trigger trigger_job_5d69fc3870b446d0a1f56a793b799ae3')

    return PostgresQueueBase

当我尝试基本模型时，

from sqlalchemy import Column, create_engine, INTEGER, TEXT
from sqlalchemy.orm import declarative_base

# IMPORTANT: inherit both a declarative base AND the postgres queue base
Base = declarative_base()
PostgresQueueBase = postgres_queue_base(schema='public')


# create custom queue model
class MyQueue(Base, PostgresQueueBase):

    # optional custom table name (by default it is "queue")
    __tablename__ = 'demo_queue'

    # custom columns
    operation = Column(TEXT)
    project_id = Column(INTEGER)


# create table in database
# change connection string accordingly!
engine = create_engine('postgresql://username:password@localhost:5432/postgres')
Base.metadata.create_all(bind=engine)

EVENT INSTRUMENT CLASS
Registering after_create event

我看不到“创建事件之后”打印出来

Answer 1

抱歉，我终于弄清楚了...该表已经存在，因此事件从未触发。另外，上面的代码在事件中存在一些错误（我无法测试它们，因为它们没有被执行），并且 table_args 中的复合索引以某种方式获得名称“”“NULLS FIRST””。我使用哈希以获得更好的名称并避免字符限制或转义问题。

import hashlib
from sqlalchemy import (BIGINT, Column, func, Index, nullslast,
                        nullsfirst, SMALLINT, TIMESTAMP)
from sqlalchemy.orm import declared_attr, declarative_mixin
from sqlalchemy.event import listens_for


# this is the function that returns the base model
def postgres_queue_base(schema:str='public', tz_aware:bool=True, use_trigger:bool=True) -> 'PostgresQueueBase':

    @declarative_mixin  # this is only for MyPy, it does not modify anything
    class PostgresQueueBase:

        __tablename__ = 'queue'

        @declared_attr
        def __table_args__(cls):
            # to prevent any problems such as escaping, SQL injection or limit of characters I'll just md5 the table name for the index
            md5 = hashlib.md5(cls.__tablename__.encode('utf-8')).hexdigest()
            return (Index(f'queue_prio_ix_{md5}', nullsfirst(cls.schedule_at), nullslast(cls.dequeued_at),
                          postgresql_where=(cls.dequeued_at == None)),
                    {'schema':schema})

        id = Column('id', BIGINT, primary_key=True)
        internal_mapping = Column('internal_mapping', BIGINT, nullable=False)
        enqueued_at = Column('enqueued_at', TIMESTAMP(timezone=tz_aware), nullable=False, server_default=func.now())
        dequeued_at = Column('dequeued_at', TIMESTAMP(timezone=tz_aware))
        expected_at = Column(TIMESTAMP(timezone=tz_aware))
        schedule_at = Column(TIMESTAMP(timezone=tz_aware))
        status = Column(SMALLINT, index=True)

    if use_trigger:
        @listens_for(PostgresQueueBase, "instrument_class", propagate=True)
        def class_instrument(mapper, class_):
            if mapper.local_table is not None:
                create_trigger_event(table=mapper.local_table)

        def create_trigger_event(table):
            @listens_for(table, "after_create")
            def create_trigger(target, connection, **kw):
                print('Create trigger')

    return PostgresQueueBase