zlwuandCursor
2905dc3d5f
新增 bx_data_scope 与 BxUserContext.to_user_info，列表/我的单据 AND scope，待办审批不加 scope，审批操作 skip_scope。 Co-authored-by: Cursor <cursoragent@cursor.com>
77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
"""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,
|
|
)
|