from fastapi import APIRouter, Depends, HTTPException, status, Request from sqlalchemy.orm import Session from app import models, schemas from app.database import get_db from app.utils.auth import get_password_hash from typing import List router = APIRouter( prefix="/users", tags=["users"], responses={404: {"description": "Not found"}}, ) @router.post("/", response_model=schemas.UserResponse, status_code=status.HTTP_201_CREATED) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): """ Register a new user """ # Check if username exists db_user_by_username = db.query(models.User).filter(models.User.username == user.username).first() if db_user_by_username: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Username already registered" ) # Check if email exists db_user_by_email = db.query(models.User).filter(models.User.email == user.email).first() if db_user_by_email: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered" ) # Create new user hashed_password = get_password_hash(user.password) db_user = models.User( email=user.email, username=user.username, hashed_password=hashed_password ) db.add(db_user) db.commit() db.refresh(db_user) return db_user @router.get("/me", response_model=schemas.UserResponse) def read_users_me(request: Request): """ Get current user information """ return request.state.user @router.get("/", response_model=List[schemas.UserResponse]) def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): """ Get list of users """ users = db.query(models.User).offset(skip).limit(limit).all() return users @router.get("/{user_id}", response_model=schemas.UserResponse) def read_user(user_id: int, db: Session = Depends(get_db)): """ Get a specific user by ID """ db_user = db.query(models.User).filter(models.User.id == user_id).first() if db_user is None: raise HTTPException(status_code=404, detail="User not found") return db_user