from datetime import timedelta 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.core.security import create_access_token from app.api.deps import get_db from app.schemas.user import UserCreate, User from app.schemas.token import Token from app.crud import crud_user from app.core.logging import logger router = APIRouter() @router.post("/login", response_model=Token) async def login( db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends() ): """ OAuth2 compatible token login, get an access token for future requests. """ user = crud_user.authenticate( db, email=form_data.username, password=form_data.password ) if not user: logger.warning(f"Failed login attempt for user: {form_data.username}") raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Bearer"}, ) elif 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 ) logger.info(f"User {user.id} logged in") return {"access_token": access_token, "token_type": "bearer"} @router.post("/register", response_model=User, status_code=status.HTTP_201_CREATED) async def register(user_in: UserCreate, db: Session = Depends(get_db)): """ Register a new user. """ # Check if the user already exists user = crud_user.get_by_email(db, email=user_in.email) if user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="A user with this email already exists", ) # Create new user user = crud_user.create(db, obj_in=user_in) logger.info(f"New user registered: {user.id}") return user