shanxia2.0/backend/app/services/bx/base.py

"""报销模块 Service 基类 — 租户隔离 + UUID + 软删除。"""
from __future__ import annotations

from typing import TYPE_CHECKING, Any, List, Optional, Sequence, Tuple, Type, TypeVar

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
from app.services.bx_data_scope import apply_bx_scope, assert_bx_row_in_scope

if TYPE_CHECKING:
    from app.core.bx_auth import BxUserContext

M = TypeVar("M")


class BxCrudService:
    @staticmethod
    def get_by_id(
        model: Type[M],
        db: Session,
        id: str,
        tenant_no: str,
        *,
        user: Optional["BxUserContext"] = None,
        skip_scope: bool = False,
    ) -> Optional[M]:
        query = db.query(model).filter(model.id == id, model.tenant_no == tenant_no)
        if hasattr(model, "del_flag"):
            query = query.filter(model.del_flag == 0)
        row = query.first()
        if row and user and not skip_scope:
            assert_bx_row_in_scope(db, row, model, user, tenant_no)
        return row

    @staticmethod
    def get_paged(
        model: Type[M],
        db: Session,
        tenant_no: str,
        page: int,
        page_size: int,
        filters: Optional[List[Any]] = None,
        *,
        user: Optional["BxUserContext"] = None,
        skip_scope: bool = False,
    ) -> Tuple[Sequence[M], int]:
        query = db.query(model).filter(model.tenant_no == tenant_no)
        if hasattr(model, "del_flag"):
            query = query.filter(model.del_flag == 0)
        if filters:
            query = query.filter(*filters)
        if user and not skip_scope:
            query = apply_bx_scope(query, model, user, db, tenant_no)
        total = query.count()
        if hasattr(model, "sort_order"):
            query = query.order_by(model.sort_order.asc(), model.create_time.desc())
        else:
            query = query.order_by(model.create_time.desc())
        rows = query.offset((page - 1) * page_size).limit(page_size).all()
        return rows, total

    @staticmethod
    def create(
        model: Type[M],
        db: Session,
        obj_in: dict,
        *,
        tenant_no: str,
        user_id: str,
        dept_id: str,
    ) -> M:
        data = dict(obj_in)
        data.setdefault("id", generate_uuid())
        data["tenant_no"] = tenant_no
        data["create_by"] = user_id
        data["create_dept_id"] = dept_id or ""
        obj = model(**data)
        db.add(obj)
        db.flush()
        return obj

    @staticmethod
    def update(
        model: Type[M],
        db: Session,
        id: str,
        tenant_no: str,
        obj_in: dict,
        *,
        user_id: str,
        user: Optional["BxUserContext"] = None,
        skip_scope: bool = False,
    ) -> Optional[M]:
        obj = BxCrudService.get_by_id(
            model, db, id, tenant_no, user=user, skip_scope=skip_scope
        )
        if not obj:
            return None
        for key, value in obj_in.items():
            setattr(obj, key, value)
        if hasattr(obj, "update_by"):
            obj.update_by = user_id
        db.flush()
        return obj

    @staticmethod
    def soft_delete(
        model: Type[M],
        db: Session,
        id: str,
        tenant_no: str,
        *,
        user_id: str = "",
        user: Optional["BxUserContext"] = None,
        skip_scope: bool = False,
    ) -> bool:
        obj = BxCrudService.get_by_id(
            model, db, id, tenant_no, user=user, skip_scope=skip_scope
        )
        if not obj:
            return False
        if user:
            assert_row_deletable_by_creator(obj, user_info_from_context(user))
        elif user_id:
            assert_row_deletable_by_creator(obj, {"user_id": user_id})
        obj.del_flag = 1
        if hasattr(obj, "update_by") and user_id:
            obj.update_by = user_id
        db.flush()
        return True