from datetime import datetime from sqlalchemy import Column, BigInteger, String, event, TIMESTAMP, func from sqlalchemy.ext.declarative import declared_attr from utils.common import camel_to_snake from database import Base from id_generator import options, generator # https://github.com/yitter/IdGenerator/tree/master/Python options = options.IdGeneratorOptions(worker_id=23) idgen = generator.DefaultIdGenerator() idgen.set_id_generator(options) # 第一层基类:包含ID class IdBase(Base): __abstract__ = True id = Column(BigInteger, primary_key=True, index=True) @declared_attr def __tablename__(cls): # 自动把数据库实体类名驼峰转为数据库表名下划线 return camel_to_snake(cls.__name__) # 自动填充id @event.listens_for(IdBase, 'before_insert', propagate=True) def before_insert_listener(mapper, connection, target): if target.id is None: target.id = idgen.next_id() # 第二层基类:包含ID和审计字段 class AuditBase(IdBase): __abstract__ = True create_time = Column(TIMESTAMP, nullable=False, default=datetime.now) create_by = Column(String(255), nullable=True) update_time = Column(TIMESTAMP, nullable=False, default=datetime.now, onupdate=datetime.now) update_by = Column(String(255), nullable=True)