66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
"""
|
|
数据库模型基类
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Dict
|
|
from sqlalchemy import Column, DateTime, Integer
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.sql import func
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
class BaseModel(Base):
|
|
"""抽象基类，为所有模型提供通用字段"""
|
|
__abstract__ = True
|
|
|
|
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
|
|
created_time = Column(DateTime(timezone=True),
|
|
server_default=func.now(),
|
|
nullable=False,
|
|
comment="创建时间")
|
|
|
|
def to_dict(self, exclude: list = None) -> Dict[str, Any]:
|
|
"""
|
|
将模型实例转换为字典
|
|
|
|
Args:
|
|
exclude: 要排除的字段列表
|
|
|
|
Returns:
|
|
包含模型字段的字典
|
|
"""
|
|
exclude = exclude or []
|
|
result = {}
|
|
|
|
for column in self.__table__.columns:
|
|
if column.name in exclude:
|
|
continue
|
|
|
|
value = getattr(self, column.name)
|
|
|
|
# 处理特殊类型
|
|
if isinstance(value, datetime):
|
|
value = value.isoformat()
|
|
elif isinstance(value, bytes):
|
|
value = value.hex() if value else None
|
|
|
|
result[column.name] = value
|
|
|
|
return result
|
|
|
|
def update_from_dict(self, data: Dict[str, Any]) -> None:
|
|
"""
|
|
从字典更新模型字段
|
|
|
|
Args:
|
|
data: 包含要更新字段的字典
|
|
"""
|
|
for key, value in data.items():
|
|
if hasattr(self, key) and key != 'id':
|
|
setattr(self, key, value)
|
|
|
|
def __repr__(self) -> str:
|
|
"""模型表示"""
|
|
return f"<{self.__class__.__name__}(id={self.id})>" |