from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.api.dependencies.database import get_async_session
from app.schemas.auth import UserCreate, UserResponse, Token
from app.models.user import User
from app.auth.jwt_handler import create_access_token
from app.core.security import verify_password, get_password_hash
router = APIRouter(prefix="/auth", tags=["authentication"])
@router.post("/register", response_model=UserResponse)
async def register(
user_data: UserCreate, db: AsyncSession = Depends(get_async_session)
):
# Check if user already exists
result = await db.execute(select(User).where(User.email == user_data.email))
if result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered"
)
result = await db.execute(select(User).where(User.username == user_data.username))
if result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="Username already taken"
)
# Create new user
hashed_password = get_password_hash(user_data.password)
user = User(
email=user_data.email,
username=user_data.username,
hashed_password=hashed_password,
)
db.add(user)
await db.commit()
await db.refresh(user)
return user
@router.post("/login", response_model=Token)
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: AsyncSession = Depends(get_async_session),
):
result = await db.execute(select(User).where(User.username == form_data.username))
user = result.scalar_one_or_none()
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(data={"sub": user.username})
return {"access_token": access_token, "token_type": "bearer"}
