from pydantic import BaseModel
from uuid import UUID
class TokenSchema(BaseModel):
access_token: str
refresh_token: str
class TokenPayload(BaseModel):
sub: UUID = None
exp: int = None
from fastapi.security import OAuth2PasswordBearer
from fastapi import Depends, HTTPException, status
from models.user_model import User
from jose import jwt
from core.config import settings
from schemas.auth_schema import TokenPayload
from datetime import datetime
from pydantic import ValidationError
from services.user_service import UserService
oauth_reusavel = OAuth2PasswordBearer(
tokenUrl=f'{settings.API_V1_STR}/auth/login',
scheme_name='JWT'
)
async def get_current_user(token: str = Depends(oauth_reusavel)) -> User:
try:
payload = jwt.decode(
token,
settings.JWT_SECRET_KEY,
settings.ALGORITHM
)
token_data = TokenPayload(**payload)
if datetime.fromtimestamp(token_data.exp) < datetime.now():
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
details='Token foi expirado',
headers={'WWW-Authenticate': 'Bearer'}
)
except(jwt.JWTError, ValidationError):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
details='Erro na validação do token',
headers={'WWW-Authenticate': 'Bearer'}
)
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from typing import Any
from services.user_service import UserService
from core.security import create_access_token, create_refresh_token
from schemas.auth_schema import TokenSchema
auth_router = APIRouter()
@auth_router.post('/login',
summary='Cria Access Token e Refresh Token',
response_model=TokenSchema
)
async def login(data: OAuth2PasswordRequestForm = Depends()) -> Any:
usuario = await UserService.authenticate(
email = data.username,
password = data.password
)
if not usuario:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail='E-mail ou Senha estão incorretos'
)
return {
"access_token": create_access_token(usuario.user_id),
"refresh_token": create_refresh_token(usuario.user_id)
}