zlwuandCursor
2905dc3d5f
新增 bx_data_scope 与 BxUserContext.to_user_info，列表/我的单据 AND scope，待办审批不加 scope，审批操作 skip_scope。 Co-authored-by: Cursor <cursoragent@cursor.com>
127 lines
3.6 KiB
Python
127 lines
3.6 KiB
Python
"""报销模块 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.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,
|
|
) -> 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:
|
|
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
|
|
obj.del_flag = 1
|
|
if hasattr(obj, "update_by") and user_id:
|
|
obj.update_by = user_id
|
|
db.flush()
|
|
return True
|