shanxia2.0/backend/app/services/bx_data_scope.py

"""BX 模块数据权限辅助（create_by / create_dept_id）。"""
from __future__ import annotations

from typing import TYPE_CHECKING
from sqlalchemy.orm import Session

from app.core.data_scope import apply_data_scope, assert_row_in_data_scope

if TYPE_CHECKING:
    from app.core.bx_auth import BxUserContext


def apply_bx_scope(query, model, user: "BxUserContext", db: Session, tenant_no: str):
    """列表/统计查询应用数据权限。"""
    return apply_data_scope(query, model, user.to_user_info(), db, tenant_no)


def assert_bx_row_in_scope(
    db: Session,
    row,
    model,
    user: "BxUserContext",
    tenant_no: str,
) -> None:
    """详情/改/删前校验行级数据权限。"""
    assert_row_in_data_scope(db, row, model, user.to_user_info(), tenant_no)


def assert_bx_biz_in_scope(
    db: Session,
    biz_type: str,
    biz_id: str,
    user: "BxUserContext",
    tenant_no: str,
) -> None:
    """附件下载/删除/绑定前校验关联业务单在数据权限内。"""
    bid = (biz_id or "").strip()
    if not bid or bid == "0":
        return
    btype = (biz_type or "").strip().lower()
    if btype in ("advance", "advance_apply", "advance_issue"):
        from app.services.bx.advance import AdvanceService

        row = AdvanceService.get_by_id(db, bid, tenant_no, user=user)
        if not row:
            raise HTTPException(status_code=404, detail="关联备用金申请不存在")
        return
    if btype in ("reimburse", "reimburse_apply"):
        from app.services.bx.reimburse import ReimburseService

        row = ReimburseService.get_by_id(db, bid, tenant_no, user=user)
        if not row:
            raise HTTPException(status_code=404, detail="关联报销单不存在")
        return
    if btype == "workflow":
        from app.services.bx.workflow import WorkflowService

        row = WorkflowService.get_by_id(db, bid, tenant_no, user=user)
        if not row:
            raise HTTPException(status_code=404, detail="关联工作流不存在")


def assert_bx_attachment_biz_in_scope(
    db: Session,
    attachment,
    user: "BxUserContext",
    tenant_no: str,
) -> None:
    """按附件关联业务单校验数据权限。"""
    assert_bx_biz_in_scope(
        db,
        getattr(attachment, "biz_type", None) or "",
        getattr(attachment, "biz_id", None) or "",
        user,
        tenant_no,
    )