testapi_auth / app /routes /item.py
FEATZAFF's picture
Upload 31 files
8d6be10 verified
Raw
History Blame Contribute Delete
1.85 kB
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from app.database import get_db
from app.schemas.item import Item, ItemCreate
import app.crud as crud
router = APIRouter(
prefix="/api/items",
tags=["items"],
)
# CREATE operation
@router.post("/", response_model=Item, status_code=status.HTTP_201_CREATED)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
"""Create a new item"""
return crud.create_item(db=db, item=item)
from fastapi import Header
@router.get("/", response_model=List[Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db), api_key: str = Header(..., alias="api_key")):
"""Get all items with pagination"""
return crud.get_items(db=db, skip=skip, limit=limit, api_key=api_key)
@router.get("/{item_id}", response_model=Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
"""Get a specific item by ID"""
db_item = crud.get_item(db=db, item_id=item_id)
if db_item is None:
raise HTTPException(status_code=404, detail="Item not found")
return db_item
# UPDATE operation
@router.put("/{item_id}", response_model=Item)
def update_item(item_id: int, item: ItemCreate, db: Session = Depends(get_db)):
"""Update an existing item"""
db_item = crud.update_item(db=db, item_id=item_id, item=item)
if db_item is None:
raise HTTPException(status_code=404, detail="Item not found")
return db_item
# DELETE operation
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int, db: Session = Depends(get_db)):
"""Delete a specific item"""
success = crud.delete_item(db=db, item_id=item_id)
if not success:
raise HTTPException(status_code=404, detail="Item not found")
return None