| """ |
| API Dependencies - Authentication, DB sessions, common deps. |
| """ |
| from typing import Annotated |
|
|
| from fastapi import Depends, HTTPException, status |
| from fastapi.security import OAuth2PasswordBearer |
| from sqlalchemy.ext.asyncio import AsyncSession |
|
|
| from app.core.config import settings |
| from app.core.database import get_db |
| from app.core.security import verify_token |
| from app.models.user import User |
|
|
| oauth2_scheme = OAuth2PasswordBearer( |
| tokenUrl=f"{settings.API_V1_STR}/auth/login", |
| auto_error=False, |
| ) |
|
|
| |
| DBSession = Annotated[AsyncSession, Depends(get_db)] |
| TokenDep = Annotated[str | None, Depends(oauth2_scheme)] |
|
|
|
|
| async def get_current_user( |
| db: DBSession, |
| token: TokenDep, |
| ) -> User: |
| """Get the current authenticated user.""" |
| if not token: |
| raise HTTPException( |
| status_code=status.HTTP_401_UNAUTHORIZED, |
| detail="Not authenticated", |
| headers={"WWW-Authenticate": "Bearer"}, |
| ) |
|
|
| payload = verify_token(token, token_type="access") |
| if payload is None: |
| raise HTTPException( |
| status_code=status.HTTP_401_UNAUTHORIZED, |
| detail="Invalid or expired token", |
| headers={"WWW-Authenticate": "Bearer"}, |
| ) |
|
|
| from sqlalchemy import select |
| result = await db.execute(select(User).where(User.id == payload["sub"])) |
| user = result.scalar_one_or_none() |
|
|
| if user is None: |
| raise HTTPException( |
| status_code=status.HTTP_404_NOT_FOUND, |
| detail="User not found", |
| ) |
| if not user.is_active: |
| raise HTTPException( |
| status_code=status.HTTP_403_FORBIDDEN, |
| detail="Inactive user", |
| ) |
| return user |
|
|
|
|
| async def get_current_user_optional( |
| db: DBSession, |
| token: TokenDep, |
| ) -> User | None: |
| """Get current user if authenticated, None otherwise.""" |
| if not token: |
| return None |
| try: |
| return await get_current_user(db, token) |
| except HTTPException: |
| return None |
|
|
|
|
| async def get_current_superuser( |
| current_user: Annotated[User, Depends(get_current_user)], |
| ) -> User: |
| """Require superuser role.""" |
| if not current_user.is_superuser: |
| raise HTTPException( |
| status_code=status.HTTP_403_FORBIDDEN, |
| detail="Not enough permissions", |
| ) |
| return current_user |
|
|
|
|
| |
| CurrentUser = Annotated[User, Depends(get_current_user)] |
| OptionalUser = Annotated[User | None, Depends(get_current_user_optional)] |
| SuperUser = Annotated[User, Depends(get_current_superuser)] |
|
|
