Spaces:
No application file
No application file
| """ | |
| Authentication API routes. | |
| Handles user signup and signin endpoints. | |
| """ | |
| from fastapi import APIRouter, HTTPException, status, Depends | |
| from sqlmodel import Session, select | |
| from app.schemas.auth import ( | |
| SignupRequest, | |
| SignupResponse, | |
| SigninRequest, | |
| SigninResponse, | |
| UserResponse | |
| ) | |
| from app.models.user import User | |
| from app.core.security import hash_password | |
| from app.services.auth_service import authenticate_user | |
| from app.services.jwt_service import create_access_token | |
| from app.api.deps import get_current_user_id | |
| from app.core.database import get_db | |
| from datetime import datetime | |
| router = APIRouter(prefix="/auth", tags=["Authentication"]) | |
| async def signup(request: SignupRequest, db: Session = Depends(get_db)): | |
| """ | |
| Create a new user account. | |
| - **username**: Unique username (3-50 chars, alphanumeric + underscore) | |
| - **password**: Password (minimum 8 characters) | |
| Returns created user information. | |
| """ | |
| # Check if username already exists | |
| statement = select(User).where(User.username == request.username) | |
| existing_user = db.exec(statement).first() | |
| if existing_user: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Username already exists" | |
| ) | |
| # Validate password length | |
| if len(request.password) < 8: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Password must be at least 8 characters" | |
| ) | |
| # Hash password before storing | |
| hashed_password = hash_password(request.password) | |
| # Create user | |
| new_user = User( | |
| username=request.username, | |
| hashed_password=hashed_password, | |
| created_at=datetime.utcnow(), | |
| updated_at=datetime.utcnow() | |
| ) | |
| db.add(new_user) | |
| db.commit() | |
| db.refresh(new_user) | |
| return SignupResponse( | |
| message="Account created successfully", | |
| user=UserResponse.model_validate(new_user) | |
| ) | |
| async def signin(request: SigninRequest, db: Session = Depends(get_db)): | |
| """ | |
| Sign in with username and password. | |
| - **username**: Username | |
| - **password**: Password | |
| Returns JWT token and user information. | |
| """ | |
| # Authenticate user | |
| user = authenticate_user(db, request.username, request.password) | |
| if not user: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid username or password" | |
| ) | |
| # Generate JWT token | |
| token = create_access_token(user.id, user.username) | |
| return SigninResponse( | |
| token=token, | |
| token_type="bearer", | |
| user=UserResponse.model_validate(user) | |
| ) | |
| async def get_current_user( | |
| user_id: str = Depends(get_current_user_id), | |
| db: Session = Depends(get_db) | |
| ): | |
| """ | |
| Get current authenticated user information. | |
| Requires valid JWT token in Authorization header. | |
| """ | |
| user = db.get(User, user_id) | |
| if not user: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail="User not found" | |
| ) | |
| return UserResponse.model_validate(user) | |