/
FlysAt
/
DataFoundry
Обзор
Документация
Войти
/
FlysAt
/
DataFoundry
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
backend/src/dependencies/auth.py
62 строки
2 KB
AlexanderShmygol
refactoring
07 апр 2026, 13:55
07 апр 2026, 13:55
5cff70c
Код
Авторство
О чём код?
from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy.ext.asyncio import AsyncSession from src.core.db import get_db from src.core.security import decode_token from src.models.user import User from src.api.repositories.user import UserRepository bearer_scheme = HTTPBearer() async def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme), db: AsyncSession = Depends(get_db), ) -> User: token = credentials.credentials try: payload = decode_token(token) token_type = payload.get("type") if token_type != "access": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type", ) user_id = payload.get("sub") if user_id is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token payload", ) except ValueError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token", ) user = await UserRepository(db).get_by_id(int(user_id)) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found", ) return user def require_roles(required_roles: list[str]): async def role_checker(current_user: User = Depends(get_current_user)) -> User: user_roles = set(current_user.roles) if not user_roles.intersection(required_roles): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions", ) return current_user return role_checker