所有模型的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 /和事件监听器 - 它描述了一个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动态化(我认为这不会太难,但我至少需要弄清楚这个的全局元素).

基本上,我不希望人们必须手动在每个模型中写入此事件,因为它与这些审核列明显相关.

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

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事件.