所有模型的 SQLAlchemy 事件 after_create

我正在开发一个项目，需要对创建的每个模型进行通用定制。迄今为止我完成大部分工作的方式是通过模型继承。这是我的代码块，可以为您提供更好的想法：

app.core.dba.mixins：

class AuditExtension(MapperExtension):
    """
    AuditExtension enforces the audit column values, and ensures any interaction with
    SQLAlchemy cannot override the values
    """

    def before_insert(self, mapper, connection, instance):
        instance.created_dt = datetime.utcnow()
        instance.created_by = audit_session_user()

        instance.updated_dt = datetime.utcnow()
        instance.updated_by = audit_session_user()

    def before_update(self, mapper, connection, instance):
        # Never update the created columns
        instance.created_dt = instance.created_dt
        instance.created_by = instance.created_by

        instance.updated_dt = datetime.utcnow()
        instance.updated_by = audit_session_user()


class AuditColumns(object):

    """ Generate the column schema for simple table level auditing. """
    created_dt = Column(DateTime,
                        default=datetime.utcnow(),
                        nullable=False)
    created_by = Column(String(64),
                        #ForeignKey('operators.username', ondelete="RESTRICT"),
                        nullable=False)

    updated_dt = Column(DateTime,
                        default=datetime.utcnow(),
                        nullable=False,
                        onupdate=datetime.utcnow())

    updated_by = Column(String(64),
                        #ForeignKey('operators.username', ondelete="RESTRICT"),
                        nullable=False)

    __mapper_args__ = {
        'extension': AuditExtension()}

然后我的模型继承 AuditColumns：

class ObjectTypes(Base, AuditColumns):
    __tablename__ = 'object_types'

    id = Column(BigInteger, primary_key=True)
    name = Column(String, nullable=False, unique=True)

    def __repr__(self):
        return self.name

我的问题是；只要 Flask 应用程序和 SQLAlchemy 包含操作，我的强制审计数据解决方案就可以工作 - 这不会阻止具有数据库访问权限的任何人更新值。

因此，我现在需要在继承 AuditColumns 的每个模型上实现一个触发器。我找到了这个帖子Sqlalchemy mixins / 和事件监听器 https://stackoverflow.com/questions/12753450/sqlalchemy-mixins-and-event-listener- 它描述了 before_insert/update 的方法（我之前已经工作过），但不是“after_create”的方法。

现在，我已将其添加到我的 mixins 文件代码中（就在上面的审核代码之后：

trig_ddl = DDL("""
            CREATE TRIGGER tr_audit_columns BEFORE INSERT OR UPDATE
            ON test_table
            FOR EACH ROW EXECUTE PROCEDURE
            ss_test();
        """)

event.listen(AuditColumns, 'after_create', trig_ddl)

但是，当我运行测试用例时：

Base.metadata.drop_all(db.get_engine(app))
Base.metadata.create_all(db.get_engine(app))

我收到以下错误：

File "D:\Devel\flask-projects\sc2\app\core\dba\mixins.py", line 59, in <module>
    event.listen(AuditColumns, 'after_create', trig_ddl)
  File "D:\Devel\flask-projects\env\lib\site-packages\sqlalchemy\event.py", line 43, in listen
    (identifier, target))
sqlalchemy.exc.InvalidRequestError: No such event 'after_create' for target '<class 'app.core.dba.mixins.AuditColumns'>'

我猜这是因为它还不是一张桌子；但是我如何为表创建全局定义一个事件监听器来执行这种类型的命令？

我知道我必须使 trig_ddl 动态化（我认为这不会太难，但我至少需要弄清楚它的全局元素）。

基本上，我不希望人们在每个模型中手动编写此事件，因为它显然与这些审计列相关。

任何朝正确方向的推动都会很棒。

好吧，你需要背负这里的事件，这样你就可以得到那个Table:

@event.listens_for(AuditColumns, "instrument_class", propagate=True)
def instrument_class(mapper, class_):
    if mapper.local_table is not None:
        trigger_for_table(mapper.local_table)

def trigger_for_table(table):
    trig_ddl = DDL("""
                CREATE TRIGGER tr_%s_audit_columns BEFORE INSERT OR UPDATE
                ON %s
                FOR EACH ROW EXECUTE PROCEDURE
                ss_test();
            """ % (table.name, table.name))

    event.listen(table, 'after_create', trig_ddl)

的任何子类AuditColumms被映射，mapper.local_table已经在那里了（就像class.__table__，同样的事情），此时您应用 DDL 事件。