track-LICA/backend/app/core/deps.py

"""通用 FastAPI 依赖"""
from __future__ import annotations

from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.core.roles import ADMIN_ROLES
from app.services.auth_service import get_current_user
from app.services.data_scope_service import DataScope, resolve_data_scope


def require_roles(*roles: str):
    """生成「限定角色」依赖，避免同一个内联判断被复制到每个端点。

    用法::

        @router.get("/x")
        async def x(current_user: dict = Depends(require_admin)):
            ...

    失败一律 403 且不透露允许的角色集合（避免给探测者提供线索）。
    """
    allowed = frozenset(roles)

    async def _guard(current_user: dict = Depends(get_current_user)) -> dict:
        if (current_user or {}).get("role") not in allowed:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="当前角色无权访问该接口",
            )
        return current_user

    return _guard


# 审计日志等高权限接口复用同一实例
require_admin = require_roles(*ADMIN_ROLES)


async def get_data_scope(
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user),
) -> DataScope:
    """业务分组数据范围 —— 列表 / 统计类接口的统一入口。

    ⚠️ **硬依赖 get_current_user**：匿名请求直接 401，**不放行成空范围**。
       两个理由：
       1. 「匿名 = 空范围」会让大屏变成一个永远空白的页面，比 401 更难排查；
       2. 匿名不过滤本身就是绕过口子 —— 谁能不登录看全厂数据，分组就形同虚设。

    刻意**不加 TTL 缓存**：把人踢出组必须立即生效，任何缓存都会造成
    「已经踢了还在看」的窗口。这里是单表 + 索引扫描，成本可忽略；
    FastAPI 的依赖缓存在单请求内已经生效（同一请求多处 Depends 只解析一次）。
    """
    return await resolve_data_scope(db, current_user)