from uuid import UUID from fastapi import APIRouter, HTTPException, Security from fastapi.params import Depends from sqlalchemy.orm import Session from app.database import get_db from app.schemas.user import User, UserCreate, UserUpdate from app.models.user import User as DbUser from app.security import get_api_key router = APIRouter( prefix="/user", tags=["user (v1)"], ) @router.get(path="", response_model=list[User]) async def get_users( skip: int = 0, limit: int = 20, api_key: str = Security(get_api_key), db: Session = Depends(get_db) ): return db.query(DbUser).offset(skip).limit(limit).all() @router.get(path="/{user_id}", response_model=User) async def get_user( user_id: UUID, api_key: str = Security(get_api_key), db: Session = Depends(get_db) ): user = db.get(DbUser, user_id) if user == None: raise HTTPException(status_code=404, detail="User not found") return user @router.post(path="", status_code=201, response_model=User) async def create_user( create_user: UserCreate, api_key: str = Security(get_api_key), db: Session = Depends(get_db) ): db_user = DbUser( friendly_name=create_user.friendly_name, is_active=create_user.is_active ) db.add(db_user) db.commit() db.refresh(db_user) return db_user @router.patch(path="/{user_id}", response_model=User) async def update_user( user_id: UUID, user_update: UserUpdate, api_key: str = Security(get_api_key), db: Session = Depends(get_db) ): user = db.get(DbUser, user_id) if user is None: raise HTTPException(status_code=404, detail="User not found") for key, value in user_update.model_dump(exclude_unset=True).items(): setattr(user, key, value) db.commit() return user @router.delete(path="/{user_id}", response_model=None) async def delete_user( user_id: UUID, api_key: str = Security(get_api_key), db: Session = Depends(get_db) ): user = db.get(DbUser, user_id) if user == None: raise HTTPException(status_code=404, detail="User not found") db.delete(user) db.commit() return []