97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
from fastapi import Depends, Header
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.core.error_codes import ErrorCode
|
|
from backend.app.core.exceptions import AppException
|
|
from backend.app.db import get_db_session
|
|
from backend.app.services.audit_service import audit_service
|
|
from backend.app.services.auth_service import auth_service
|
|
from backend.app.services.config_service import config_service
|
|
from backend.app.services.customer_service import customer_service
|
|
from backend.app.services.logistics_service import logistics_service
|
|
from backend.app.services.order_service import order_service
|
|
from backend.app.services.product_service import product_service
|
|
from backend.app.services.reminder_service import reminder_service
|
|
from backend.app.services.report_service import report_service
|
|
from backend.app.services.supplier_service import supplier_service
|
|
from backend.app.services.system_service import system_service
|
|
|
|
|
|
# 统一依赖入口，后续如果切换到真实容器或数据库实现，只需要改这里。
|
|
def get_auth_service():
|
|
return auth_service
|
|
|
|
|
|
def get_current_user(
|
|
authorization: str | None = Header(default=None),
|
|
auth_service=Depends(get_auth_service),
|
|
session: Session = Depends(get_db_session),
|
|
) -> dict:
|
|
token = (authorization or "").removeprefix("Bearer").strip()
|
|
user = auth_service.get_me(token, session)
|
|
if not user:
|
|
raise AppException(code=ErrorCode.UNAUTHORIZED, message="未登录或登录失效", status_code=401)
|
|
return user
|
|
|
|
|
|
def require_roles(*role_codes: str):
|
|
def _require_roles(current_user: dict = Depends(get_current_user)) -> dict:
|
|
if role_codes and current_user.get("role_code") not in role_codes:
|
|
raise AppException(code=ErrorCode.FORBIDDEN, message="无权限访问", status_code=403)
|
|
return current_user
|
|
|
|
return _require_roles
|
|
|
|
|
|
def require_permissions(*permission_codes: str):
|
|
def _require_permissions(current_user: dict = Depends(get_current_user)) -> dict:
|
|
if not permission_codes:
|
|
return current_user
|
|
granted = set(current_user.get("permissions") or [])
|
|
missing = [code for code in permission_codes if code not in granted]
|
|
if missing:
|
|
raise AppException(code=ErrorCode.FORBIDDEN, message="缺少必要权限", status_code=403)
|
|
return current_user
|
|
|
|
return _require_permissions
|
|
|
|
|
|
def get_order_service():
|
|
return order_service
|
|
|
|
|
|
def get_customer_service():
|
|
return customer_service
|
|
|
|
|
|
def get_product_service():
|
|
return product_service
|
|
|
|
|
|
def get_logistics_service():
|
|
return logistics_service
|
|
|
|
|
|
def get_supplier_service():
|
|
return supplier_service
|
|
|
|
|
|
def get_reminder_service():
|
|
return reminder_service
|
|
|
|
|
|
def get_report_service():
|
|
return report_service
|
|
|
|
|
|
def get_config_service():
|
|
return config_service
|
|
|
|
|
|
def get_audit_service():
|
|
return audit_service
|
|
|
|
|
|
def get_system_service():
|
|
return system_service
|