"""Authentication routes.""" from datetime import timedelta from fastapi import APIRouter, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from sqlalchemy.orm import Session from passlib.context import CryptContext from app.database import get_db from app.models.user import User from app.schemas.auth import Token, LoginRequest from app.schemas.user import UserCreate, UserResponse from app.config import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") router = APIRouter(prefix="/auth", tags=["auth"]) def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify password.""" return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: """Hash password.""" return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: timedelta | None = None): """Create access token.""" from jose import jwt to_encode = data.copy() expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) @router.post("/register", response_model=UserResponse) def register(user: UserCreate, db: Session = Depends(get_db)): """Register new user.""" db_user = db.query(User).filter(User.email == user.email).first() if db_user: raise HTTPException(status_code=400, detail="Email already registered") hashed_password = get_password_hash(user.password) db_user = User( email=user.email, username=user.username, hashed_password=hashed_password, full_name=user.full_name ) db.add(db_user) db.commit() db.refresh(db_user) return db_user @router.post("/login", response_model=Token) def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)): """Login user.""" user = db.query(User).filter(User.username == form_data.username).first() 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" ) access_token = create_access_token( data={"sub": str(user.id)}, expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) ) return {"access_token": access_token, "token_type": "bearer"}