zlwuandCursor
f5ab023ab5
为 apply_data_scope 增加 owner_field 与行级校验，补充 DbUserContext.to_user_info() 及单元测试，并在用户/角色/字典下拉与用户详情接口生效。 Co-authored-by: Cursor <cursoragent@cursor.com>
82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
"""地磅模块认证适配：门户 JWT 用户 → db 业务上下文。"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Optional
|
|
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.auth import get_current_user_by_token
|
|
from app.core.database import get_db
|
|
from app.core.menu_guard import ADMIN_ROLES, require_admin_or_super_admin
|
|
|
|
optional_oauth2_scheme = OAuth2PasswordBearer(
|
|
tokenUrl="/api/public/login",
|
|
auto_error=False,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class DbUserContext:
|
|
"""db Router/Service 使用的当前用户上下文。"""
|
|
|
|
id: str
|
|
name: str
|
|
dept_id: str
|
|
role: str
|
|
tenant_no: str
|
|
|
|
def to_tenant_user_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"user_id": self.id,
|
|
"role": self.role,
|
|
"tenant_no": self.tenant_no,
|
|
}
|
|
|
|
def to_user_info(self) -> Dict[str, Any]:
|
|
"""与门户 JWT / apply_data_scope 所需的 user_info 字段对齐。"""
|
|
return {
|
|
"user_id": self.id,
|
|
"role": self.role,
|
|
"tenant_no": self.tenant_no,
|
|
"dept_id": self.dept_id,
|
|
"region_id": self.dept_id,
|
|
}
|
|
|
|
|
|
def get_db_current_user(
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
) -> DbUserContext:
|
|
return DbUserContext(
|
|
id=(user.get("user_id") or "").strip(),
|
|
name=(user.get("real_name") or user.get("username") or "").strip(),
|
|
dept_id=(user.get("region_id") or "").strip(),
|
|
role=(user.get("role") or "").strip(),
|
|
tenant_no=(user.get("tenant_no") or "").strip(),
|
|
)
|
|
|
|
|
|
async def get_db_current_user_optional(
|
|
token: Optional[str] = Depends(optional_oauth2_scheme),
|
|
db: Session = Depends(get_db),
|
|
) -> DbUserContext | None:
|
|
"""有 Bearer 且为 admin/super_admin 时返回上下文，否则 None（供三方 H5 无登录调用）。"""
|
|
if not token:
|
|
return None
|
|
try:
|
|
user = await get_current_user_by_token(token, db)
|
|
except HTTPException:
|
|
return None
|
|
role = user.get("role")
|
|
if role not in ADMIN_ROLES:
|
|
return None
|
|
return DbUserContext(
|
|
id=(user.get("user_id") or "").strip(),
|
|
name=(user.get("real_name") or user.get("username") or "").strip(),
|
|
dept_id=(user.get("region_id") or "").strip(),
|
|
role=(role or "").strip(),
|
|
tenant_no=(user.get("tenant_no") or "").strip(),
|
|
)
|