from datetime import timedelta from typing import Any from fastapi import APIRouter, Depends, HTTPException, status from fastapi.security import OAuth2PasswordRequestForm from sqlalchemy.orm import Session from app.core.config import settings from app.db.session import get_db from app.services.auth import authenticate_user from app.services.security import create_access_token from app.schemas.auth import Token router = APIRouter() @router.post("/token", response_model=Token) async def login_for_access_token( form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db) ) -> Any: """ OAuth2 compatible token login, get an access token for future requests. """ user = authenticate_user(db, email=form_data.username, password=form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Bearer"}, ) # Check if user is active if not user.is_active: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user", ) # Create access token access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( subject=user.id, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"}