Spaces:
Running
Running
File size: 1,431 Bytes
d23039a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | from __future__ import annotations
from fastapi import Depends, Header, HTTPException, status
from sqlalchemy.orm import Session
from db.database import get_db
from db.models import User
from services.auth_service import decode_token, get_user
def _extract_bearer(authorization: str | None) -> str | None:
if not authorization:
return None
parts = authorization.split()
if len(parts) != 2 or parts[0].lower() != "bearer":
return None
return parts[1]
def get_current_user(
authorization: str | None = Header(default=None),
db: Session = Depends(get_db),
) -> User:
token = _extract_bearer(authorization)
if not token:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Missing bearer token")
payload = decode_token(token)
if not payload or "sub" not in payload:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid or expired token")
user = get_user(db, int(payload["sub"]))
if not user:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "User not found")
return user
def optional_current_user(
authorization: str | None = Header(default=None),
db: Session = Depends(get_db),
) -> User | None:
token = _extract_bearer(authorization)
if not token:
return None
payload = decode_token(token)
if not payload or "sub" not in payload:
return None
return get_user(db, int(payload["sub"]))
|