57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
PerToolBox Server - 依赖注入
|
|
Copyright (C) 2024 Sea Network Technology Studio
|
|
Author: Canglan <admin@sea-studio.top>
|
|
License: AGPL v3
|
|
"""
|
|
|
|
from typing import Annotated, Optional
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .database import get_db
|
|
from .models import User
|
|
from .utils.security import decode_access_token
|
|
from .utils.logger import logger
|
|
|
|
security = HTTPBearer()
|
|
|
|
DbDependency = Annotated[Session, Depends(get_db)]
|
|
|
|
async def get_current_user(
|
|
db: DbDependency,
|
|
credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)]
|
|
) -> Optional[User]:
|
|
token = credentials.credentials
|
|
payload = decode_access_token(token)
|
|
|
|
if not payload:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="无效的 token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
user_id = payload.get("sub")
|
|
if not user_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="无效的 token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
user = db.query(User).filter(User.id == int(user_id)).first()
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="用户不存在",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
return user
|
|
|
|
CurrentUserDependency = Annotated[User, Depends(get_current_user)]
|
|
OptionalCurrentUserDependency = Annotated[Optional[User], Depends(get_current_user)] |