from typing import Any, List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.api import deps from app.models.item import Item from app.schemas.item import Item as ItemSchema from app.schemas.item import ItemCreate, ItemUpdate router = APIRouter() @router.get("/", response_model=List[ItemSchema]) def read_items( db: Session = Depends(deps.get_db), skip: int = 0, limit: int = 100, ) -> Any: """ Retrieve items. """ items = db.query(Item).offset(skip).limit(limit).all() return items @router.post("/", response_model=ItemSchema) def create_item( *, db: Session = Depends(deps.get_db), item_in: ItemCreate, ) -> Any: """ Create new item. """ item = Item(title=item_in.title, description=item_in.description) db.add(item) db.commit() db.refresh(item) return item @router.get("/{item_id}", response_model=ItemSchema) def read_item( *, db: Session = Depends(deps.get_db), item_id: int, ) -> Any: """ Get item by ID. """ item = db.query(Item).filter(Item.id == item_id).first() if not item: raise HTTPException(status_code=404, detail="Item not found") return item @router.put("/{item_id}", response_model=ItemSchema) def update_item( *, db: Session = Depends(deps.get_db), item_id: int, item_in: ItemUpdate, ) -> Any: """ Update an item. """ item = db.query(Item).filter(Item.id == item_id).first() if not item: raise HTTPException(status_code=404, detail="Item not found") update_data = item_in.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(item, field, value) db.add(item) db.commit() db.refresh(item) return item @router.delete( "/{item_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None ) def delete_item( *, db: Session = Depends(deps.get_db), item_id: int, ) -> Any: """ Delete an item. """ item = db.query(Item).filter(Item.id == item_id).first() if not item: raise HTTPException(status_code=404, detail="Item not found") db.delete(item) db.commit() return None