119 lines
3.5 KiB
Python
119 lines
3.5 KiB
Python
"""删除权限：仅创建人可删（super_admin 例外）。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, Iterable, Optional, Union
|
|
|
|
from fastapi import HTTPException
|
|
|
|
RowLike = Union[Any, Dict[str, Any]]
|
|
|
|
|
|
def normalize_user_info(
|
|
user_info: Optional[Dict[str, Any]] = None,
|
|
*,
|
|
user_id: Optional[str] = None,
|
|
actor_name: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
info: Dict[str, Any] = dict(user_info or {})
|
|
uid = (user_id or info.get("user_id") or info.get("id") or "").strip()
|
|
if uid:
|
|
info["user_id"] = uid
|
|
name = (actor_name or info.get("real_name") or info.get("username") or info.get("userName") or info.get("name") or "").strip()
|
|
if name:
|
|
info["real_name"] = name
|
|
return info
|
|
|
|
|
|
def user_info_from_context(ctx_or_dict: Any) -> Dict[str, Any]:
|
|
if isinstance(ctx_or_dict, dict):
|
|
return normalize_user_info(ctx_or_dict)
|
|
info: Dict[str, Any] = {}
|
|
if hasattr(ctx_or_dict, "to_user_info"):
|
|
info.update(ctx_or_dict.to_user_info())
|
|
name = (getattr(ctx_or_dict, "name", None) or "").strip()
|
|
if name:
|
|
info["real_name"] = name
|
|
return normalize_user_info(info)
|
|
|
|
|
|
def is_super_admin(user_info: Optional[Dict[str, Any]]) -> bool:
|
|
if not user_info:
|
|
return False
|
|
return (user_info.get("role") or "").strip() == "super_admin"
|
|
|
|
|
|
def row_creator_id(row: RowLike) -> str:
|
|
if row is None:
|
|
return ""
|
|
if isinstance(row, dict):
|
|
return (row.get("create_by") or row.get("createBy") or "").strip()
|
|
return (getattr(row, "create_by", None) or getattr(row, "createBy", None) or "").strip()
|
|
|
|
|
|
def _actor_ids(user_info: Optional[Dict[str, Any]]) -> set[str]:
|
|
if not user_info:
|
|
return set()
|
|
ids: set[str] = set()
|
|
for key in ("user_id", "id"):
|
|
val = (user_info.get(key) or "").strip()
|
|
if val:
|
|
ids.add(val)
|
|
return ids
|
|
|
|
|
|
def _actor_names(user_info: Optional[Dict[str, Any]]) -> set[str]:
|
|
if not user_info:
|
|
return set()
|
|
names: set[str] = set()
|
|
for key in ("real_name", "username", "userName", "name"):
|
|
val = (user_info.get(key) or "").strip()
|
|
if val:
|
|
names.add(val)
|
|
return names
|
|
|
|
|
|
def can_delete_by_creator(
|
|
row: RowLike,
|
|
user_info: Optional[Dict[str, Any]] = None,
|
|
*,
|
|
user_id: Optional[str] = None,
|
|
actor_name: Optional[str] = None,
|
|
) -> bool:
|
|
info = normalize_user_info(user_info, user_id=user_id, actor_name=actor_name)
|
|
if is_super_admin(info):
|
|
return True
|
|
creator = row_creator_id(row)
|
|
if not creator:
|
|
return False
|
|
if creator in _actor_ids(info):
|
|
return True
|
|
if creator in _actor_names(info):
|
|
return True
|
|
return False
|
|
|
|
|
|
def assert_row_deletable_by_creator(
|
|
row: RowLike,
|
|
user_info: Optional[Dict[str, Any]] = None,
|
|
*,
|
|
user_id: Optional[str] = None,
|
|
actor_name: Optional[str] = None,
|
|
message: str = "仅创建人可删除",
|
|
) -> None:
|
|
if can_delete_by_creator(row, user_info, user_id=user_id, actor_name=actor_name):
|
|
return
|
|
raise HTTPException(status_code=403, detail=message)
|
|
|
|
|
|
def assert_rows_deletable_by_creator(
|
|
rows: Iterable[RowLike],
|
|
user_info: Optional[Dict[str, Any]] = None,
|
|
*,
|
|
user_id: Optional[str] = None,
|
|
actor_name: Optional[str] = None,
|
|
message: str = "仅创建人可删除",
|
|
) -> None:
|
|
info = normalize_user_info(user_info, user_id=user_id, actor_name=actor_name)
|
|
for row in rows:
|
|
assert_row_deletable_by_creator(row, info, message=message)
|