36 lines
1.3 KiB
Python
36 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, Header, HTTPException, status
|
|
|
|
from app.core.database import get_connection
|
|
from app.core.schemas import UserProfile
|
|
from app.services import repository
|
|
|
|
|
|
def get_bearer_token(authorization: Annotated[str | None, Header()] = None) -> str:
|
|
if not authorization or not authorization.startswith("Bearer "):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing_token")
|
|
return authorization.replace("Bearer ", "", 1).strip()
|
|
|
|
|
|
def get_current_user(token: Annotated[str, Depends(get_bearer_token)]) -> UserProfile:
|
|
with get_connection() as connection:
|
|
row = repository.get_user_by_token(connection, token)
|
|
if row is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid_token")
|
|
return repository.profile_from_row(row)
|
|
|
|
|
|
def require_student(user: Annotated[UserProfile, Depends(get_current_user)]) -> UserProfile:
|
|
if user.role != "student":
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="student_only")
|
|
return user
|
|
|
|
|
|
def require_admin(user: Annotated[UserProfile, Depends(get_current_user)]) -> UserProfile:
|
|
if user.role != "admin":
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="admin_only")
|
|
return user
|