"""Authentication endpoints for API v1.""" from fastapi import APIRouter, Depends, HTTPException, status, Request from fastapi.security import OAuth2PasswordRequestForm from sqlalchemy.orm import Session from typing import List from backend.models.user import User from backend import auth from backend.schemas.user import UserCreate, UserResponse from backend.schemas.token import Token from backend.database import get_db from backend.core.config import settings from slowapi import Limiter from slowapi.util import get_remote_address router = APIRouter(prefix="/api/auth", tags=["authentication"]) limiter = Limiter(key_func=get_remote_address) @router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED) @limiter.limit(f"{settings.RATE_LIMIT_TIMES}/{settings.RATE_LIMIT_SECONDS}s") def register(request: Request, user: UserCreate, db: Session = Depends(get_db)): """Register a new user.""" db_user = db.query(User).filter(User.username == user.username).first() if db_user: raise HTTPException(status_code=400, detail="Username already registered") hashed_password = auth.get_password_hash(user.password) new_user = User(username=user.username, hashed_password=hashed_password) db.add(new_user) db.commit() db.refresh(new_user) return new_user @router.post("/token", response_model=Token) @limiter.limit(f"{settings.RATE_LIMIT_TIMES}/{settings.RATE_LIMIT_SECONDS}s") def login_for_access_token( request: Request, form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db) ): """Login and get access token.""" user = db.query(User).filter(User.username == form_data.username).first() if not user or not auth.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"}, ) from datetime import timedelta access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) access_token = auth.create_access_token( data={"sub": user.username}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @router.get("/me", response_model=UserResponse) def read_users_me(current_user: User = Depends(auth.get_current_user)): """Get current user information.""" return current_user