所有模型的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」。

現在，我已經添加到了我的混入文件的代碼（後直奔我上面的審計代碼：

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