DjangoAdmin_FastAPI_EleVue/shared/base_model.py

# +======================================================================
# | 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)))