sqlalchemy 监听所有实体插入以及更新事件

这边使用的是flask+dependency-injecter+sqlalchemy，有一个公共类，想插入或者更新的时候对公共类某些字段进行统一操作

这个是公共类：包括一些基础字段，所有的实体都会继承这个类 

"""Models module."""from datetime import datetime
from sqlalchemy import Column, String, Integer,DateTime#, Booleanfrom sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()class CommonEntity(Base):__abstract__ = Truecreated_dt = Column(DateTime)created_by = Column(String)updated_dt = Column(DateTime)updated_by = Column(String)version = Column(Integer)

然后是db的配置

"""Database module."""from contextlib import contextmanager, AbstractContextManager
from typing import Callablefrom sqlalchemy import create_engine, orm,event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
from datetime import datetimefrom main.services.common_service.db_access.domain.common_field_entity import CommonEntity
Base = declarative_base()class DatabaseConfig:def __init__(self, db_url: str) -> None:self._engine = create_engine(db_url, echo=True)self._session_factory = orm.scoped_session(orm.sessionmaker(autocommit=False,autoflush=False,expire_on_commit=False,bind=self._engine,),)def create_database(self) -> None:Base.metadata.create_all(self._engine)@contextmanagerdef session(self) -> Callable[..., AbstractContextManager[Session]]:session: Session = self._session_factory()try:yield sessionexcept Exception:session.rollback()raiseelse:if session._transaction.is_active:session.commit()session.close()@event.listens_for(CommonEntity, 'before_insert', propagate=True)def before_insert_listener(self, mapper, target):# 在创建时自动更新 created_dt，versiontarget.created_dt = datetime.now()target.created_by = 'Damien'target.version = [e-mail]_for(CommonEntity, 'before_update', propagate=True)def before_update_listener(self, mapper, target):# 在更新时自动更新 updated_dt，versiontarget.updated_dt = datetime.now()target.updated_by = 'Damien'target.version += 1

 重要的就是@event.listens_for，这里。监听公共类，所有继承了它的实体的插入更新都会被监听到，然后就是对里面的字段进行统一的操作就行了，不用再每次插入或者更新进行手动修改了