diff --git a/main.py b/main.py index aeaeec7..5f6ca0d 100644 --- a/main.py +++ b/main.py @@ -1,10 +1,10 @@ from fastapi import FastAPI, HTTPException -from pydantic import BaseModel +from pydantic import BaseModel, Field import sqlite3 from datetime import datetime from typing import Optional, List -app = FastAPI() +app = FastAPI(title="Shopping List API", version="0.1.0") DB_PATH = "shopping.db" @@ -49,24 +49,35 @@ def init_db(): def startup(): init_db() +# Request models class Product(BaseModel): - name: str - sku: Optional[str] = None + name: str = Field(..., min_length=1, description="Product name") + sku: Optional[str] = Field(None, description="Optional SKU") +class ListModel(BaseModel): + name: str = Field(..., min_length=1, description="List name") + +class ListItemCreate(BaseModel): + product_id: int = Field(..., gt=0, description="ID of the product") + quantity: int = Field(1, ge=1, description="Quantity to add (>=1)") + +class ListItemUpdate(BaseModel): + quantity: int = Field(..., ge=1, description="New quantity") + +# Response models class ProductResponse(Product): id: int created_at: datetime -class ListModel(BaseModel): - name: str + class Config: + from_attributes = True class ListResponse(ListModel): id: int created_at: datetime -class ListItemCreate(BaseModel): - product_id: int - quantity: int = 1 + class Config: + from_attributes = True class ListItemResponse(BaseModel): id: int @@ -77,37 +88,44 @@ class ListItemResponse(BaseModel): product_name: Optional[str] = None product_sku: Optional[str] = None + class Config: + from_attributes = True + class ListWithItems(ListResponse): items: List[ListItemResponse] = [] -# Products +# Endpoints -@app.post("/products", response_model=ProductResponse, status_code=201) +@app.get("/", tags=["meta"]) +def read_root(): + return {"message": "Shopping List API", "version": "0.1.0"} + +@app.post("/products", response_model=ProductResponse, status_code=201, tags=["products"]) def create_product(product: Product): conn = get_db() c = conn.cursor() try: c.execute( "INSERT INTO products (name, sku) VALUES (?, ?)", - (product.name, product.sku) + (product.name.strip(), product.sku) ) conn.commit() pid = c.lastrowid - except sqlite3.IntegrityError as e: + except sqlite3.IntegrityError: conn.close() raise HTTPException(status_code=400, detail="Product name must be unique") row = conn.execute("SELECT * FROM products WHERE id = ?", (pid,)).fetchone() conn.close() return ProductResponse(**dict(row)) -@app.get("/products", response_model=List[ProductResponse]) +@app.get("/products", response_model=List[ProductResponse], tags=["products"]) def list_products(): conn = get_db() rows = conn.execute("SELECT * FROM products ORDER BY id").fetchall() conn.close() return [ProductResponse(**dict(row)) for row in rows] -@app.get("/products/{id}", response_model=ProductResponse) +@app.get("/products/{id}", response_model=ProductResponse, tags=["products"]) def get_product(id: int): conn = get_db() row = conn.execute("SELECT * FROM products WHERE id = ?", (id,)).fetchone() @@ -116,39 +134,37 @@ def get_product(id: int): raise HTTPException(status_code=404, detail="Product not found") return ProductResponse(**dict(row)) -@app.delete("/products/{id}") +@app.delete("/products/{id}", tags=["products"]) def delete_product(id: int): conn = get_db() - cur = conn.cursor() - cur.execute("DELETE FROM products WHERE id = ?", (id,)) - if cur.rowcount == 0: + c = conn.cursor() + c.execute("DELETE FROM products WHERE id = ?", (id,)) + if c.rowcount == 0: conn.close() raise HTTPException(status_code=404, detail="Product not found") conn.commit() conn.close() return {"deleted": id} -# Lists - -@app.post("/lists", response_model=ListResponse, status_code=201) +@app.post("/lists", response_model=ListResponse, status_code=201, tags=["lists"]) def create_list(lst: ListModel): conn = get_db() c = conn.cursor() - c.execute("INSERT INTO lists (name) VALUES (?)", (lst.name,)) + c.execute("INSERT INTO lists (name) VALUES (?)", (lst.name.strip(),)) conn.commit() lid = c.lastrowid row = conn.execute("SELECT * FROM lists WHERE id = ?", (lid,)).fetchone() conn.close() return ListResponse(**dict(row)) -@app.get("/lists", response_model=List[ListResponse]) +@app.get("/lists", response_model=List[ListResponse], tags=["lists"]) def list_lists(): conn = get_db() rows = conn.execute("SELECT * FROM lists ORDER BY id").fetchall() conn.close() return [ListResponse(**dict(row)) for row in rows] -@app.get("/lists/{id}", response_model=ListWithItems) +@app.get("/lists/{id}", response_model=ListWithItems, tags=["lists"]) def get_list(id: int): conn = get_db() lst_row = conn.execute("SELECT * FROM lists WHERE id = ?", (id,)).fetchone() @@ -166,19 +182,18 @@ def get_list(id: int): conn.close() items = [] for row in items_rows: - d = dict(row) items.append(ListItemResponse( - id=d['id'], - list_id=d['list_id'], - product_id=d['product_id'], - quantity=d['quantity'], - added_at=d['added_at'], - product_name=d.get('product_name'), - product_sku=d.get('product_sku') + id=row['id'], + list_id=row['list_id'], + product_id=row['product_id'], + quantity=row['quantity'], + added_at=row['added_at'], + product_name=row.get('product_name'), + product_sku=row.get('product_sku') )) return ListWithItems(**dict(lst_row), items=items) -@app.delete("/lists/{id}") +@app.delete("/lists/{id}", tags=["lists"]) def delete_list(id: int): conn = get_db() c = conn.cursor() @@ -190,18 +205,16 @@ def delete_list(id: int): conn.close() return {"deleted": id} -# List Items - -@app.post("/lists/{list_id}/items", response_model=ListItemResponse, status_code=201) +@app.post("/lists/{list_id}/items", response_model=ListItemResponse, status_code=201, tags=["items"]) def add_item(list_id: int, item: ListItemCreate): conn = get_db() # Verify product exists - prod = conn.execute("SELECT * FROM products WHERE id = ?", (item.product_id,)).fetchone() + prod = conn.execute("SELECT id FROM products WHERE id = ?", (item.product_id,)).fetchone() if not prod: conn.close() raise HTTPException(status_code=404, detail="Product not found") # Verify list exists - lst = conn.execute("SELECT * FROM lists WHERE id = ?", (list_id,)).fetchone() + lst = conn.execute("SELECT id FROM lists WHERE id = ?", (list_id,)).fetchone() if not lst: conn.close() raise HTTPException(status_code=404, detail="List not found") @@ -219,34 +232,33 @@ def add_item(list_id: int, item: ListItemCreate): WHERE li.id = ? """, (iid,)).fetchone() conn.close() - d = dict(row) return ListItemResponse( - id=d['id'], - list_id=d['list_id'], - product_id=d['product_id'], - quantity=d['quantity'], - added_at=d['added_at'], - product_name=d.get('product_name'), - product_sku=d.get('product_sku') + id=row['id'], + list_id=row['list_id'], + product_id=row['product_id'], + quantity=row['quantity'], + added_at=row['added_at'], + product_name=row.get('product_name'), + product_sku=row.get('product_sku') ) -@app.patch("/lists/{list_id}/items/{item_id}") -def update_item_quantity(list_id: int, item_id: int, quantity: int): +@app.patch("/lists/{list_id}/items/{item_id}", tags=["items"]) +def update_item_quantity(list_id: int, item_id: int, payload: ListItemUpdate): + quantity = payload.quantity conn = get_db() - # Verify item belongs to list - cur = conn.cursor() - cur.execute( + c = conn.cursor() + c.execute( "UPDATE list_items SET quantity = ? WHERE id = ? AND list_id = ?", (quantity, item_id, list_id) ) - if cur.rowcount == 0: + if c.rowcount == 0: conn.close() raise HTTPException(status_code=404, detail="Item not found in list") conn.commit() conn.close() return {"list_id": list_id, "item_id": item_id, "quantity": quantity} -@app.delete("/lists/{list_id}/items/{item_id}") +@app.delete("/lists/{list_id}/items/{item_id}", tags=["items"]) def remove_item(list_id: int, item_id: int): conn = get_db() c = conn.cursor() @@ -258,11 +270,11 @@ def remove_item(list_id: int, item_id: int): conn.close() return {"list_id": list_id, "item_id": item_id} -@app.get("/lists/{list_id}/items", response_model=List[ListItemResponse]) +@app.get("/lists/{list_id}/items", response_model=List[ListItemResponse], tags=["items"]) def list_items(list_id: int): conn = get_db() # Verify list exists - lst = conn.execute("SELECT * FROM lists WHERE id = ?", (list_id,)).fetchone() + lst = conn.execute("SELECT id FROM lists WHERE id = ?", (list_id,)).fetchone() if not lst: conn.close() raise HTTPException(status_code=404, detail="List not found") @@ -286,7 +298,3 @@ def list_items(list_id: int): ) for row in rows ] - -@app.get("/") -def read_root(): - return {"message": "Shopping List API"}