Implement core API endpoints (WI2)\n\n- Add full CRUD for Products (create, list, get, delete)\n- Add CRUD for Lists (create, list, get with items, delete)\n- Add List Items management (add, update quantity, delete, list)\n- Proper error handling with HTTPException\n- Use sqlite3 with row_factory for dict-like rows\n- Response models with Pydantic\n\nOperation: shopping-list-api-2026-04-05\nWI2: Implement core API endpoints

2026-04-05 23:32:19 +00:00
parent 8db5262f46
commit b0ee2a2a5b
1 changed files with 237 additions and 4 deletions
--- a/main.py
+++ b/main.py
@@ -2,13 +2,19 @@ from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
 import sqlite3
 from datetime import datetime
+from typing import Optional, List

 app = FastAPI()

 DB_PATH = "shopping.db"

-def init_db():
+def get_db():
    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+def init_db():
+    conn = get_db()
    c = conn.cursor()
    c.execute("""
        CREATE TABLE IF NOT EXISTS products (
@@ -45,15 +51,242 @@ def startup():

 class Product(BaseModel):
    name: str
-    sku: str | None = None
+    sku: Optional[str] = None

-class List(BaseModel):
+class ProductResponse(Product):
+    id: int
+    created_at: datetime
+
+class ListModel(BaseModel):
    name: str

-class ListItem(BaseModel):
+class ListResponse(ListModel):
+    id: int
+    created_at: datetime
+
+class ListItemCreate(BaseModel):
    product_id: int
    quantity: int = 1

+class ListItemResponse(BaseModel):
+    id: int
+    list_id: int
+    product_id: int
+    quantity: int
+    added_at: datetime
+    product_name: Optional[str] = None
+    product_sku: Optional[str] = None
+
+class ListWithItems(ListResponse):
+    items: List[ListItemResponse] = []
+
+# Products
+
+@app.post("/products", response_model=ProductResponse, status_code=201)
+def create_product(product: Product):
+    conn = get_db()
+    c = conn.cursor()
+    try:
+        c.execute(
+            "INSERT INTO products (name, sku) VALUES (?, ?)",
+            (product.name, product.sku)
+        )
+        conn.commit()
+        pid = c.lastrowid
+    except sqlite3.IntegrityError as e:
+        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])
+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)
+def get_product(id: int):
+    conn = get_db()
+    row = conn.execute("SELECT * FROM products WHERE id = ?", (id,)).fetchone()
+    conn.close()
+    if not row:
+        raise HTTPException(status_code=404, detail="Product not found")
+    return ProductResponse(**dict(row))
+
+@app.delete("/products/{id}")
+def delete_product(id: int):
+    conn = get_db()
+    cur = conn.cursor()
+    cur.execute("DELETE FROM products WHERE id = ?", (id,))
+    if cur.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)
+def create_list(lst: ListModel):
+    conn = get_db()
+    c = conn.cursor()
+    c.execute("INSERT INTO lists (name) VALUES (?)", (lst.name,))
+    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])
+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)
+def get_list(id: int):
+    conn = get_db()
+    lst_row = conn.execute("SELECT * FROM lists WHERE id = ?", (id,)).fetchone()
+    if not lst_row:
+        conn.close()
+        raise HTTPException(status_code=404, detail="List not found")
+    items_rows = conn.execute("""
+        SELECT li.id, li.list_id, li.product_id, li.quantity, li.added_at,
+               p.name AS product_name, p.sku AS product_sku
+        FROM list_items li
+        JOIN products p ON li.product_id = p.id
+        WHERE li.list_id = ?
+        ORDER BY li.id
+    """, (id,)).fetchall()
+    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')
+        ))
+    return ListWithItems(**dict(lst_row), items=items)
+
+@app.delete("/lists/{id}")
+def delete_list(id: int):
+    conn = get_db()
+    c = conn.cursor()
+    c.execute("DELETE FROM lists WHERE id = ?", (id,))
+    if c.rowcount == 0:
+        conn.close()
+        raise HTTPException(status_code=404, detail="List not found")
+    conn.commit()
+    conn.close()
+    return {"deleted": id}
+
+# List Items
+
+@app.post("/lists/{list_id}/items", response_model=ListItemResponse, status_code=201)
+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()
+    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()
+    if not lst:
+        conn.close()
+        raise HTTPException(status_code=404, detail="List not found")
+    c = conn.cursor()
+    c.execute(
+        "INSERT INTO list_items (list_id, product_id, quantity) VALUES (?, ?, ?)",
+        (list_id, item.product_id, item.quantity)
+    )
+    conn.commit()
+    iid = c.lastrowid
+    row = conn.execute("""
+        SELECT li.*, p.name AS product_name, p.sku AS product_sku
+        FROM list_items li
+        JOIN products p ON li.product_id = p.id
+        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')
+    )
+
+@app.patch("/lists/{list_id}/items/{item_id}")
+def update_item_quantity(list_id: int, item_id: int, quantity: int):
+    conn = get_db()
+    # Verify item belongs to list
+    cur = conn.cursor()
+    cur.execute(
+        "UPDATE list_items SET quantity = ? WHERE id = ? AND list_id = ?",
+        (quantity, item_id, list_id)
+    )
+    if cur.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}")
+def remove_item(list_id: int, item_id: int):
+    conn = get_db()
+    c = conn.cursor()
+    c.execute("DELETE FROM list_items WHERE id = ? AND list_id = ?", (item_id, list_id))
+    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}
+
+@app.get("/lists/{list_id}/items", response_model=List[ListItemResponse])
+def list_items(list_id: int):
+    conn = get_db()
+    # Verify list exists
+    lst = conn.execute("SELECT * FROM lists WHERE id = ?", (list_id,)).fetchone()
+    if not lst:
+        conn.close()
+        raise HTTPException(status_code=404, detail="List not found")
+    rows = conn.execute("""
+        SELECT li.*, p.name AS product_name, p.sku AS product_sku
+        FROM list_items li
+        JOIN products p ON li.product_id = p.id
+        WHERE li.list_id = ?
+        ORDER BY li.id
+    """, (list_id,)).fetchall()
+    conn.close()
+    return [
+        ListItemResponse(
+            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')
+        )
+        for row in rows
+    ]
+
@app.get("/")
 def read_root():
    return {"message": "Shopping List API"}