当前位置: 首页 > news >正文

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 = 1@event.listens_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,这里。监听公共类,所有继承了它的实体的插入更新都会被监听到,然后就是对里面的字段进行统一的操作就行了,不用再每次插入或者更新进行手动修改了

http://www.lryc.cn/news/279147.html

相关文章:

  • go怎么结束很多个协程呢
  • springboot 项目访问静态资源遇到的问题,WebMvcConfigurer和WebMvcConfigurationSupport
  • Nginx配置负载均衡实例
  • 【算法题】50. Pow(x, n)
  • K8S动态PV
  • 逆变器2(原理框图)
  • ERA5合集,使用ERA5得到GNSS站点的温度,气压,水汽压,Tm和PWV合集,可以求五个参数
  • c#让三个线程按照顺序执行
  • AWS Directory Service 开启ldaps
  • Seata 以 Nacos 为注册中心启动
  • Unity填坑-灯光烘焙相关
  • 【python】TCP测速程序
  • 新书速览|从零开始大模型开发与微调:基于PyTorch与ChatGLM
  • 边缘计算:连接实时数据的力量与未来发展之路
  • ZooKeeper 实战(四) Curator Watch事件监听
  • Spring Boot 构建工具插件
  • Java集成消息队列Kafka
  • 第十四章JSON
  • 0_项目git地址——正点原子minifly与crazyflie
  • php 字符串常用函数
  • Android基于Matrix绘制PaintDrawable设置BitmapShader,以手指触点为中心显示原图像圆图,Kotlin(2)
  • FlinkOnYarn 监控 flink任务
  • C++内存管理机制(侯捷)笔记1
  • 【论文阅读】Non-blocking Lazy Schema Changes in Multi-Version
  • Rust 最新版1.75.0升级记
  • 使用 KubeSphere 与极狐GitLab 打造云原生持续交付系统
  • EasyExcel的追加写入(新增POI、CSV)
  • JetBrains 开发工具——免费教育许可申请流程
  • 打造高性价比小程序,轻松降低成本
  • mysql 索引优化查询