308 lines
9.7 KiB
Python
308 lines
9.7 KiB
Python
from contextlib import asynccontextmanager
|
|
from datetime import datetime
|
|
import os
|
|
import sqlite3
|
|
from typing import List, Optional
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
init_db()
|
|
yield
|
|
|
|
|
|
app = FastAPI(title="Shopping List API", version="0.1.0", lifespan=lifespan)
|
|
|
|
|
|
def get_db_path():
|
|
return os.environ.get("DB_PATH", "shopping.db")
|
|
|
|
|
|
def get_db():
|
|
conn = sqlite3.connect(get_db_path())
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
return conn
|
|
|
|
def init_db():
|
|
conn = get_db()
|
|
c = conn.cursor()
|
|
c.execute("""
|
|
CREATE TABLE IF NOT EXISTS products (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
sku TEXT,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
c.execute("""
|
|
CREATE TABLE IF NOT EXISTS lists (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
c.execute("""
|
|
CREATE TABLE IF NOT EXISTS list_items (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
list_id INTEGER NOT NULL,
|
|
product_id INTEGER NOT NULL,
|
|
quantity INTEGER NOT NULL DEFAULT 1,
|
|
added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (list_id) REFERENCES lists(id) ON DELETE CASCADE,
|
|
FOREIGN KEY (product_id) REFERENCES products(id)
|
|
)
|
|
""")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
# Request models
|
|
class Product(BaseModel):
|
|
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):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
created_at: datetime
|
|
|
|
class ListResponse(ListModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
created_at: datetime
|
|
|
|
class ListItemResponse(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
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] = []
|
|
|
|
# Endpoints
|
|
|
|
@app.get("/", tags=["meta"])
|
|
def read_root():
|
|
return {"message": "Shopping List API"}
|
|
|
|
@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.strip(), product.sku)
|
|
)
|
|
conn.commit()
|
|
pid = c.lastrowid
|
|
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], 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, tags=["products"])
|
|
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}", tags=["products"])
|
|
def delete_product(id: int):
|
|
conn = get_db()
|
|
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}
|
|
|
|
@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.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], 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, tags=["lists"])
|
|
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:
|
|
items.append(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['product_name'],
|
|
product_sku=row['product_sku']
|
|
))
|
|
return ListWithItems(**dict(lst_row), items=items)
|
|
|
|
@app.delete("/lists/{id}", tags=["lists"])
|
|
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}
|
|
|
|
@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 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 id 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()
|
|
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['product_name'],
|
|
product_sku=row['product_sku']
|
|
)
|
|
|
|
@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()
|
|
c = conn.cursor()
|
|
c.execute(
|
|
"UPDATE list_items SET quantity = ? WHERE id = ? AND list_id = ?",
|
|
(quantity, 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, "quantity": quantity}
|
|
|
|
@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()
|
|
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], tags=["items"])
|
|
def list_items(list_id: int):
|
|
conn = get_db()
|
|
# Verify list exists
|
|
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")
|
|
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['product_name'],
|
|
product_sku=row['product_sku']
|
|
)
|
|
for row in rows
|
|
]
|