/
electro_bayan
/
area_safety
Обзор
Документация
Войти
/
electro_bayan
/
area_safety
Код
Запросы
0
Задачи
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
src/api/dependencies.py
55 строк
2 KB
Artem
Initial
21 сен 2024, 09:38
21 сен 2024, 09:38
ded25b8
Код
Авторство
О чём код?
from typing import AsyncGenerator from sqlalchemy.ext.asyncio import AsyncSession from core.db.session import engine from typing import Annotated from fastapi import Depends from fastapi.security import OAuth2PasswordBearer from config import settings import jwt from pydantic import ValidationError from fastapi.exceptions import HTTPException from fastapi import status from models.user import User from core.fastapi_utils.schemas.token import TokenPayload reusable_oauth2 = OAuth2PasswordBearer( tokenUrl=f"/login/access-token" ) async def get_db(): async with AsyncSession(engine) as session: yield session SessionDep = Annotated[AsyncSession, Depends(get_db)] TokenDep = Annotated[str, Depends(reusable_oauth2)] async def get_current_user(session: SessionDep, token: TokenDep) -> User: try: payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) token_data = TokenPayload(**payload) except (jwt.InvalidTokenError, ValidationError): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Could not validate user") user = await session.get(User, token_data.sub) if not user: raise HTTPException(status_code=404, detail="User not found") if not user.is_active: raise HTTPException(status_code=400, detail="Inactive user") return user CurrentUserDep = Annotated[User, Depends(get_current_user)] async def get_current_superuser(current_user: CurrentUserDep) -> User: if not current_user.is_superuser: raise HTTPException(status_code=403, detail="Forbidden") return current_user CurrentSuperuserDep = Annotated[User, Depends(get_current_superuser)]