from typing import List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.db.session import get_db from app.db.models.user import User from app.schemas.user import UserCreate, UserResponse, UserUpdate from app.core.security import get_password_hash router = APIRouter() @router.post("/users/", response_model=UserResponse, status_code=status.HTTP_201_CREATED) def create_user( user_in: UserCreate, db: Session = Depends(get_db) ): """ Create a new user. """ # Check if user with given email exists user = db.query(User).filter(User.email == user_in.email).first() if user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered" ) # Check if username is taken user = db.query(User).filter(User.username == user_in.username).first() if user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Username already taken" ) # Create new user hashed_password = get_password_hash(user_in.password) user = User( email=user_in.email, username=user_in.username, hashed_password=hashed_password, is_active=user_in.is_active, is_superuser=user_in.is_superuser ) db.add(user) db.commit() db.refresh(user) return user @router.get("/users/", response_model=List[UserResponse]) def get_users( skip: int = 0, limit: int = 100, db: Session = Depends(get_db) ): """ Get all users. """ users = db.query(User).offset(skip).limit(limit).all() return users @router.get("/users/{user_id}", response_model=UserResponse) def get_user( user_id: int, db: Session = Depends(get_db) ): """ Get a specific user by ID. """ user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) return user @router.patch("/users/{user_id}", response_model=UserResponse) def update_user( user_id: int, user_in: UserUpdate, db: Session = Depends(get_db) ): """ Update a user. """ user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) # Update user fields if provided update_data = user_in.dict(exclude_unset=True) # Hash the password if it's being updated if "password" in update_data: update_data["hashed_password"] = get_password_hash(update_data.pop("password")) for field, value in update_data.items(): setattr(user, field, value) db.add(user) db.commit() db.refresh(user) return user @router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None) def delete_user( user_id: int, db: Session = Depends(get_db) ): """ Delete a user. """ user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) db.delete(user) db.commit() return None