feat(phase2): shared/ 共享层迁移

- shared/: enums/constants/pagination/base_model/base_db/base_repository/base_service
- apps/constants 与 apps/models/base_model、base_db 改为垫片
- 新增 pagination 单元测试

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
2026-08-07 10:03:38 +08:00
co-authored by Claude
parent db4ee85db5
commit bde742c7a8
11 changed files with 674 additions and 290 deletions
+149
View File
@@ -0,0 +1,149 @@
# +======================================================================
# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者,助力企业发展 ]
# +----------------------------------------------------------------------
# | Copyright (c) 2021-2026 一米阳光团队 版权所有
# | 开源协议: Apache-2.0
# +======================================================================
# 由 apps/models/base_model.py 迁移而来。
"""数据库模型基类:通用字段、序列化与通用查询方法。"""
import logging
from datetime import date, datetime, timezone
from typing import Optional, List, Dict, Any
from sqlalchemy import Column, Integer, DATETIME, and_
from core import R
from core.database import Base, get_db_session
from shared.constants import SERIALIZE_EXCLUDE_FIELDS
from utils.request import parse_id_list
logger = logging.getLogger(__name__)
_UTC_NOW = lambda: datetime.now(tz=timezone.utc)
class base_model(Base):
"""
数据库模型基类
所有业务模型应继承此类,自动获得通用字段和公共方法。
提供的功能:
1. 通用字段:id、创建人、创建时间、更新人、更新时间、软删除标识
2. 序列化方法:to_dict()
3. 查询方法:get_by_id()、get_by_ids()、get_all_active()、exists()
4. 删除方法:batch_delete_r()
"""
__abstract__ = True
# ============================================================
# 通用字段定义
# ============================================================
id = Column(Integer, primary_key=True, autoincrement=True, comment="主键ID")
create_user = Column(Integer, default=0, comment="创建人")
create_time = Column(DATETIME, default=_UTC_NOW, comment="创建时间")
update_user = Column(Integer, default=0, comment="更新人")
update_time = Column(DATETIME, default=_UTC_NOW, onupdate=_UTC_NOW, comment="更新时间")
is_delete = Column(Integer, default=0, comment="删除标识:0-正常 1-已删除")
# ============================================================
# 序列化方法
# ============================================================
def to_dict(self) -> Dict[str, Any]:
"""
将模型实例转换为字典
自动排除敏感字段(is_delete、password、salt)。
datetime 类型格式化为字符串,UTC 时间转为本地时间便于前端展示。
"""
result = {}
for column in self.__table__.columns:
if column.name in SERIALIZE_EXCLUDE_FIELDS:
continue
value = getattr(self, column.name)
if isinstance(value, datetime):
if value.tzinfo is not None:
# UTC 时间转为本地时间
value = value.astimezone()
value = value.strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(value, date):
# 纯日期类型(如生日)无时分秒,单独格式化,避免 JSON 序列化失败
value = value.strftime('%Y-%m-%d')
result[column.name] = value
return result
# ============================================================
# 通用查询方法
# ============================================================
@classmethod
def get_by_id(cls, id: Optional[int]) -> Optional['base_model']:
"""根据ID获取未删除的记录,不存在返回 None"""
if id is None or id <= 0:
return None
db = get_db_session()
return db.query(cls).filter(cls.id == id, cls.is_delete == 0).first()
@classmethod
def get_by_ids(cls, ids: List[int]) -> List['base_model']:
"""根据ID列表批量获取未删除的记录"""
if not ids:
return []
valid_ids = [i for i in ids if i and i > 0]
if not valid_ids:
return []
db = get_db_session()
return db.query(cls).filter(cls.id.in_(valid_ids), cls.is_delete == 0).all()
@classmethod
def get_all_active(cls) -> List['base_model']:
"""获取所有未删除的记录"""
db = get_db_session()
return db.query(cls).filter(cls.is_delete == 0).all()
@classmethod
def exists(cls, id: Optional[int]) -> bool:
"""检查未删除记录是否存在"""
if id is None or id <= 0:
return False
db = get_db_session()
return db.query(cls).filter(cls.id == id, cls.is_delete == 0).count() > 0
# ============================================================
# 删除方法
# ============================================================
@classmethod
def batch_delete_r(cls, ids_str: Optional[str]):
"""
批量软删除并返回R对象
封装ID校验、解析、软删除的完整流程。
Args:
ids_str: 逗号分隔的ID字符串,如 "1,2,3"
Returns:
R: 操作结果
"""
if not ids_str:
return R.failed("记录ID不存在")
id_list = parse_id_list(ids_str)
if not id_list:
return R.failed("记录ID不存在")
db = get_db_session()
records = db.query(cls).filter(and_(cls.id.in_(id_list), cls.is_delete == 0)).all()
if len(records) != len(id_list):
return R.failed("记录不存在")
for record in records:
record.is_delete = 1
return R.ok(msg="本次共删除{0}条数据".format(len(records)))