203 lines
5.8 KiB
Python
203 lines
5.8 KiB
Python
"""数据权限过滤（V0.3：基于角色 data_scope 的查询级行级过滤）
|
||
|
||
data_scope 值：
|
||
1 = 全部数据
|
||
2 = 本部门数据
|
||
3 = 本部门及以下数据
|
||
4 = 仅本人数据
|
||
5 = 自定义（sys_role_data_scope_detail）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import List, Set, Tuple
|
||
|
||
from sqlalchemy import or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.models.models import (
|
||
SysDept,
|
||
SysRole,
|
||
SysRoleDataScopeDetail,
|
||
SysUser,
|
||
SysUserRole,
|
||
)
|
||
|
||
|
||
def get_user_dept_id(db: Session, user_id: str) -> str | None:
|
||
"""从 SysUser 读取用户的 dept_id。"""
|
||
row = (
|
||
db.query(SysUser.dept_id)
|
||
.filter(SysUser.id == user_id, SysUser.del_flag == 0)
|
||
.first()
|
||
)
|
||
return row[0] if row and row[0] else None
|
||
|
||
|
||
def _get_child_dept_ids(db: Session, dept_id: str, tenant_no: str) -> List[str]:
|
||
"""递归获取 dept_id 及其所有子部门 ID（含自身）。"""
|
||
result: List[str] = [dept_id]
|
||
# 用循环逐层扩展，避免深度递归栈溢出
|
||
queue: List[str] = [dept_id]
|
||
while queue:
|
||
parents = queue
|
||
queue = []
|
||
children = (
|
||
db.query(SysDept.id)
|
||
.filter(
|
||
SysDept.parent_id.in_(parents),
|
||
SysDept.del_flag == 0,
|
||
SysDept.tenant_no == tenant_no,
|
||
)
|
||
.all()
|
||
)
|
||
for child in children:
|
||
cid = child[0]
|
||
if cid not in result:
|
||
result.append(cid)
|
||
queue.append(cid)
|
||
return result
|
||
|
||
|
||
def _resolve_user_role_scopes(
|
||
db: Session, user_id: str, tenant_no: str
|
||
) -> List[dict]:
|
||
"""获取用户所有角色的数据权限配置。
|
||
|
||
返回: [{"data_scope": str, "details": [SysRoleDataScopeDetail, ...]}]
|
||
"""
|
||
roles = (
|
||
db.query(SysRole)
|
||
.join(SysUserRole, SysUserRole.role_id == SysRole.id)
|
||
.filter(
|
||
SysUserRole.user_id == user_id,
|
||
SysUserRole.del_flag == 0,
|
||
SysUserRole.tenant_no == tenant_no,
|
||
SysRole.del_flag == 0,
|
||
SysRole.status == "1",
|
||
SysRole.tenant_no == tenant_no,
|
||
)
|
||
.all()
|
||
)
|
||
if not roles:
|
||
return []
|
||
|
||
result: List[dict] = []
|
||
role_ids = [r.id for r in roles]
|
||
|
||
# 批量查所有自定义权限明细
|
||
all_details = (
|
||
db.query(SysRoleDataScopeDetail)
|
||
.filter(
|
||
SysRoleDataScopeDetail.role_id.in_(role_ids),
|
||
SysRoleDataScopeDetail.del_flag == 0,
|
||
SysRoleDataScopeDetail.tenant_no == tenant_no,
|
||
)
|
||
.all()
|
||
)
|
||
details_by_role: dict[str, list] = {}
|
||
for d in all_details:
|
||
details_by_role.setdefault(d.role_id, []).append(d)
|
||
|
||
for role in roles:
|
||
result.append(
|
||
{
|
||
"data_scope": role.data_scope or "1",
|
||
"details": details_by_role.get(role.id, []),
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def apply_data_scope(query, model, user_info: dict, db: Session, tenant_no: str = None):
|
||
"""根据用户角色的数据权限过滤查询。
|
||
|
||
Args:
|
||
query: SQLAlchemy query 对象
|
||
model: 查询的主模型类（需要有 create_dept_id / create_by 字段）
|
||
user_info: 当前用户信息字典，需含 role / user_id / dept_id 等
|
||
db: 数据库 Session
|
||
tenant_no: 租户编号，默认使用 settings.DEFAULT_TENANT_NO
|
||
|
||
Returns:
|
||
应用数据权限过滤后的 query
|
||
"""
|
||
# 1. 无用户信息 → 不过滤
|
||
if not user_info:
|
||
return query
|
||
|
||
# 2. super_admin 直接放行
|
||
if user_info.get("role") == "super_admin":
|
||
return query
|
||
|
||
# 3. 获取 user_id
|
||
user_id = user_info.get("user_id")
|
||
if not user_id:
|
||
return query
|
||
|
||
# 4. 确定 tenant_no
|
||
effective_tenant = tenant_no or settings.DEFAULT_TENANT_NO
|
||
|
||
# 5. 检查 model 是否有可过滤字段
|
||
has_dept = hasattr(model, "create_dept_id")
|
||
has_user = hasattr(model, "create_by")
|
||
if not has_dept and not has_user:
|
||
return query
|
||
|
||
# 6. 获取用户所有角色的数据权限
|
||
role_scopes = _resolve_user_role_scopes(db, user_id, effective_tenant)
|
||
if not role_scopes:
|
||
return query.filter(False)
|
||
|
||
# 7. 任一角色拥有全部数据权限 → 放行（union 语义）
|
||
if any(rs["data_scope"] == "1" for rs in role_scopes):
|
||
return query
|
||
|
||
# 8. 收集所有角色的 dept_ids 和 user_ids（union 合并）
|
||
dept_ids: Set[str] = set()
|
||
user_ids: Set[str] = set()
|
||
|
||
# 获取用户部门
|
||
user_dept_id = user_info.get("dept_id") or user_info.get("region_id") or get_user_dept_id(db, user_id)
|
||
|
||
for rs in role_scopes:
|
||
scope = rs["data_scope"]
|
||
|
||
if scope == "2":
|
||
# 本部门
|
||
if user_dept_id:
|
||
dept_ids.add(user_dept_id)
|
||
|
||
elif scope == "3":
|
||
# 本部门及以下
|
||
if user_dept_id:
|
||
dept_ids.update(_get_child_dept_ids(db, user_dept_id, effective_tenant))
|
||
|
||
elif scope == "4":
|
||
# 仅本人
|
||
user_ids.add(user_id)
|
||
|
||
elif scope == "5":
|
||
# 自定义
|
||
for detail in rs["details"]:
|
||
val = (detail.scope_value or "").strip()
|
||
if not val:
|
||
continue
|
||
if detail.scope_type == "dept":
|
||
dept_ids.add(val)
|
||
elif detail.scope_type == "user":
|
||
user_ids.add(val)
|
||
|
||
# 9. 构建 OR 条件并应用
|
||
conditions = []
|
||
if dept_ids and has_dept:
|
||
conditions.append(model.create_dept_id.in_(dept_ids))
|
||
if user_ids and has_user:
|
||
conditions.append(model.create_by.in_(user_ids))
|
||
|
||
if not conditions:
|
||
# 没有任何可匹配条件 → 返回空结果
|
||
return query.filter(False)
|
||
|
||
return query.filter(or_(*conditions))
|