Hera Zhao
d19183923c
Extracted from oil project — business logic removed, auth/db/deploy infrastructure generalized with APP_NAME placeholders. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
from fastapi import Request, Depends, HTTPException
|
|
|
|
from backend.database import get_db
|
|
|
|
ANON_USER = {"id": None, "role": "viewer", "username": "anonymous", "display_name": "匿名"}
|
|
|
|
|
|
def get_current_user(request: Request):
|
|
"""Extract user from Bearer token. Returns anonymous if no/invalid token."""
|
|
token = request.headers.get("Authorization", "").removeprefix("Bearer ").strip()
|
|
if not token:
|
|
return ANON_USER
|
|
conn = get_db()
|
|
user = conn.execute(
|
|
"SELECT id, username, role, display_name, password FROM users WHERE token = ?",
|
|
(token,),
|
|
).fetchone()
|
|
conn.close()
|
|
if not user:
|
|
return ANON_USER
|
|
return dict(user)
|
|
|
|
|
|
def require_role(*roles):
|
|
"""Dependency that checks the user has one of the given roles."""
|
|
def checker(user=Depends(get_current_user)):
|
|
if user["role"] not in roles:
|
|
raise HTTPException(403, "权限不足")
|
|
return user
|
|
return checker
|
|
|
|
|
|
def require_login(user=Depends(get_current_user)):
|
|
"""Dependency that requires any authenticated user."""
|
|
if user["id"] is None:
|
|
raise HTTPException(401, "请先登录")
|
|
return user
|