为所有模型创建后的SQLAlchemy事件

Question

我正在做一个项目，这个项目需要对创建的每个模型进行通用定制。到目前为止，我所做的大部分工作都是通过模型继承来完成的。下面是我的代码块，让您更好地了解：

在app.core.dba软件混音器：

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：

^{pr2}$

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

因此，我现在需要在继承AuditColumns的每个模型上实现一个触发器。我发现了这篇文章Sqlalchemy mixins / and event listener-它描述了一个用于before插入/更新的方法（我以前也有过这样的工作），但没有描述“after\u create”。在

现在，我已经将此添加到mixin文件代码中（在我上面的审计代码之后：

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'>'

我猜这是因为它还不是一个表；但是我如何为一个表create全局定义一个事件侦听器来执行这种类型的命令？在

我知道我必须使trig&ddl成为动态的（我认为这不会太难，但我至少需要弄清楚其中的全局元素）。在

基本上，我不想让人们在每个模型中手动编写这个事件，而这个事件显然与这些审计列有关联。在

任何朝着正确的方向推进都会很好。在

Answer 1

好吧，你需要把事件放在这里，这样你就可以得到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将已经存在（与{}相同），此时应用DDL事件。在