Cheng Zhou
27 lines
878 B
Python
27 lines
878 B
Python
from typing import Annotated, AsyncGenerator
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from pydantic import ValidationError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.security import decode_access_token, oauth2_scheme
|
|
from app.db.session import SessionLocal
|
|
from app.schemas.user import TokenPayload
|
|
|
|
|
|
async def get_db() -> AsyncGenerator[AsyncSession, None]:
|
|
async with SessionLocal() as session:
|
|
yield session
|
|
|
|
|
|
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> TokenPayload:
|
|
payload = decode_access_token(token)
|
|
try:
|
|
return TokenPayload(**payload)
|
|
except ValidationError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Could not validate credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
) from exc
|