58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
from sqlalchemy import Column, BigInteger, String, DateTime, event
|
||
from sqlalchemy.ext.declarative import declared_attr
|
||
|
||
from config.auth import context_sub
|
||
from config.database import Base
|
||
from datetime import datetime
|
||
|
||
from utils.common import camel_to_snake
|
||
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(DateTime, nullable=True, default=datetime.now)
|
||
create_by = Column(String(255), nullable=True)
|
||
update_time = Column(DateTime, nullable=True, default=datetime.now, onupdate=datetime.now)
|
||
update_by = Column(String(255), nullable=True)
|
||
|
||
|
||
@event.listens_for(AuditBase, 'before_insert', propagate=True)
|
||
def before_insert(mapper, connection, target):
|
||
value = context_sub.get(None)
|
||
if value is not None:
|
||
target.create_by = value
|
||
|
||
|
||
@event.listens_for(AuditBase, 'before_update', propagate=True)
|
||
def before_update(mapper, connection, target):
|
||
value = context_sub.get(None)
|
||
if value is not None:
|
||
target.update_by = value
|