4

我正在处理一个需要对创建的每个模型进行通用自定义的项目。迄今为止，我完成大部分工作的方式是模型继承。这是我的代码块，可让您更好地了解：

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

我的问题是；只要操作包含在烧瓶应用程序和 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'>'

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

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

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

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

4

1 回答 1

6

那么你需要在这里捎带这些事件，这样你就可以做到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 事件。

于 2013-07-11T00:56:24.160 回答