Polish API implementation\n\n- Add FastAPI metadata (title, version)\n- Use Field for validation (min_length, ge)\n- Add response model Config (from_attributes)\n- Strip name inputs\n- Better error messages\n- Tags for OpenAPI grouping\n\nOperation: shopping-list-api-2026-04-05\nWI4: Finalization and documentation

This commit is contained in:
Marcus A.
2026-04-05 23:45:38 +00:00
parent 08633862d7
commit dd6b76e665

128
main.py
View File

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