python - 所有模型的 SQLAlchemy 事件 after_create

Question

我正在从事一个项目，该项目需要对创建的每个模型进行通用定制。迄今为止，我完成大部分工作的方式是通过模型继承。这是我的代码块，可以给您一个更好的主意:

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

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

我知道我必须使 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 将已经存在(class.__table__ 也将存在，同样的事情)，您在该点应用 DDL 事件。