DjangoAdmin_FastAPI_EleVue/shared/base_repository.py

# +======================================================================
# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ]
# +----------------------------------------------------------------------
# | Copyright (c) 2021-2026 一米阳光团队 版权所有
# | 开源协议: Apache-2.0
# +======================================================================

"""通用 Repository 基类：封装各模型的数据访问层公共操作。

各模块 repository.py 继承本类并提供自定义查询，service 通过 repository 访问数据库。
过渡期与 base_model 的 classmethod 双轨并存，全部模块迁移完成后收敛。
"""

from typing import Generic, Optional, TypeVar

from core.database import get_db_session
from shared.constants import PAGE_SIZE
from utils.request import parse_id_list

M = TypeVar("M")


class BaseRepository(Generic[M]):
    """模型数据访问基类。"""

    # 软删除列名，JobLogRepository 覆盖为 "del_flag"
    soft_delete_col: str = "is_delete"

    def __init__(self, model_cls):
        self.model = model_cls

    @property
    def db(self):
        """当前请求的数据库会话。"""
        return get_db_session()

    def _soft_col(self):
        return getattr(self.model, self.soft_delete_col)

    # ============================================================
    # 基础查询
    # ============================================================

    def get_by_id(self, obj_id: Optional[int]) -> Optional[M]:
        """根据ID获取未删除记录（兼容字符串ID，如 URL 路径参数）。"""
        if obj_id is None or obj_id == '':
            return None
        try:
            obj_id = int(obj_id)
        except (TypeError, ValueError):
            return None
        if obj_id <= 0:
            return None
        return self.db.query(self.model).filter(
            self.model.id == obj_id, self._soft_col() == 0
        ).first()

    def get_by_ids(self, ids) -> list:
        """根据ID列表批量获取未删除记录。"""
        valid = []
        for i in ids:
            try:
                i = int(i)
            except (TypeError, ValueError):
                continue
            if i > 0:
                valid.append(i)
        if not valid:
            return []
        return self.db.query(self.model).filter(
            self.model.id.in_(valid), self._soft_col() == 0
        ).all()

    def get_all_active(self) -> list:
        """获取所有未删除记录。"""
        return self.db.query(self.model).filter(self._soft_col() == 0).all()

    def exists(self, obj_id: Optional[int]) -> bool:
        """检查未删除记录是否存在。"""
        return self.get_by_id(obj_id) is not None

    # ============================================================
    # 分页查询
    # ============================================================

    def paginate(self, page_no: int = 1, page_size: int = PAGE_SIZE, query=None):
        """
        分页查询，返回 (items, total)。自动追加软删过滤。
        传入 query 可在外部先构建过滤条件。
        """
        q = query if query is not None else self.db.query(self.model)
        q = q.filter(self._soft_col() == 0)
        total = q.count()
        items = q.limit(page_size).offset((page_no - 1) * page_size).all()
        return items, total

    # ============================================================
    # 写操作
    # ============================================================

    def create(self, **fields) -> M:
        """创建记录（不提交事务，由中间件统一管理）。"""
        obj = self.model(**fields)
        obj.save()
        return obj

    def update(self, obj_id, data: dict) -> int:
        """按ID更新，返回受影响行数。"""
        return self.db.query(self.model).filter(
            self.model.id == obj_id, self._soft_col() == 0
        ).update(data)

    def soft_delete(self, ids) -> bool:
        """
        批量软删除（等价 base_model.batch_delete_r 的语义）。

        Args:
            ids: 逗号分隔字符串或整数列表

        Returns:
            bool: 是否全部删除成功（ID 数需与匹配记录数一致）
        """
        id_list = parse_id_list(ids) if isinstance(ids, str) else [i for i in ids if i]
        if not id_list:
            return False
        records = self.db.query(self.model).filter(
            self.model.id.in_(id_list), self._soft_col() == 0
        ).all()
        if len(records) != len(id_list):
            return False
        for record in records:
            setattr(record, self.soft_delete_col, 1)
        return True

    def hard_delete(self, obj) -> bool:
        """物理删除实例。"""
        self.db.delete(obj)
        self.db.flush()
        return True

    # ============================================================
    # 自定义查询出口
    # ============================================================

    def filter(self, *criterion):
        """返回过滤后的 query，供子类组合复杂查询。"""
        return self.db.query(self.model).filter(*criterion)