v1.0.0提交

dependencies.py

@@ -0,0 +1,57 @@
+#!/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)]