255 lines
9.2 KiB
Python
255 lines
9.2 KiB
Python
# backend/app/api/shopping.py
|
|
import uuid
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from fastapi.responses import Response
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.core.database import get_session
|
|
from app.models.shopping import ShoppingList, ListItem, Product, Store
|
|
from app.schemas.shopping import (
|
|
ShoppingListCreate, ShoppingListUpdate, ShoppingListResponse,
|
|
ShoppingListDetailResponse, ListItemCreate, ListItemUpdate,
|
|
ListItemResponse, ProductResponse, StoreResponse,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _item_to_response(item: ListItem) -> ListItemResponse:
|
|
display_name = item.custom_name or (item.product.name if item.product else "Article inconnu")
|
|
return ListItemResponse(
|
|
id=item.id,
|
|
product_id=item.product_id,
|
|
custom_name=item.custom_name,
|
|
display_name=display_name,
|
|
quantity=item.quantity,
|
|
unit=item.unit,
|
|
is_checked=item.is_checked,
|
|
price_recorded=item.price_recorded,
|
|
carried_over=item.carried_over,
|
|
sort_order=item.sort_order,
|
|
)
|
|
|
|
|
|
def _list_to_response(lst: ShoppingList) -> ShoppingListResponse:
|
|
items = lst.items if lst.items is not None else []
|
|
return ShoppingListResponse(
|
|
id=lst.id,
|
|
name=lst.name,
|
|
store_id=lst.store_id,
|
|
week_date=lst.week_date,
|
|
status=lst.status,
|
|
created_at=lst.created_at,
|
|
item_count=len(items),
|
|
checked_count=sum(1 for i in items if i.is_checked),
|
|
)
|
|
|
|
|
|
# ── Stores ────────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/stores", response_model=list[StoreResponse])
|
|
async def list_stores(session: AsyncSession = Depends(get_session)):
|
|
result = await session.execute(select(Store).order_by(Store.name))
|
|
return result.scalars().all()
|
|
|
|
|
|
# ── Products ──────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/products", response_model=list[ProductResponse])
|
|
async def search_products(
|
|
q: str | None = Query(default=None),
|
|
limit: int = 30,
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
stmt = select(Product).order_by(Product.frequency_score.desc(), Product.name)
|
|
if q:
|
|
stmt = stmt.where(Product.name.ilike(f"%{q}%"))
|
|
stmt = stmt.limit(limit)
|
|
result = await session.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
# ── Lists ─────────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/lists", response_model=list[ShoppingListResponse])
|
|
async def list_shopping_lists(session: AsyncSession = Depends(get_session)):
|
|
stmt = (
|
|
select(ShoppingList)
|
|
.options(selectinload(ShoppingList.items))
|
|
.order_by(ShoppingList.created_at.desc())
|
|
)
|
|
result = await session.execute(stmt)
|
|
lists = result.scalars().all()
|
|
return [_list_to_response(lst) for lst in lists]
|
|
|
|
|
|
@router.post("/lists", response_model=ShoppingListDetailResponse, status_code=201)
|
|
async def create_shopping_list(
|
|
payload: ShoppingListCreate,
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
lst = ShoppingList(**payload.model_dump())
|
|
session.add(lst)
|
|
await session.commit()
|
|
await session.refresh(lst, ["items"])
|
|
return ShoppingListDetailResponse(
|
|
**_list_to_response(lst).model_dump(),
|
|
items=[],
|
|
)
|
|
|
|
|
|
@router.get("/lists/{list_id}", response_model=ShoppingListDetailResponse)
|
|
async def get_shopping_list(list_id: uuid.UUID, session: AsyncSession = Depends(get_session)):
|
|
stmt = (
|
|
select(ShoppingList)
|
|
.where(ShoppingList.id == list_id)
|
|
.options(selectinload(ShoppingList.items).selectinload(ListItem.product))
|
|
)
|
|
result = await session.execute(stmt)
|
|
lst = result.scalar_one_or_none()
|
|
if not lst:
|
|
raise HTTPException(404, "Liste introuvable")
|
|
sorted_items = sorted(lst.items, key=lambda i: (i.sort_order or 999, str(i.id)))
|
|
return ShoppingListDetailResponse(
|
|
**_list_to_response(lst).model_dump(),
|
|
items=[_item_to_response(i) for i in sorted_items],
|
|
)
|
|
|
|
|
|
@router.patch("/lists/{list_id}", response_model=ShoppingListDetailResponse)
|
|
async def update_shopping_list(
|
|
list_id: uuid.UUID,
|
|
payload: ShoppingListUpdate,
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
stmt = (
|
|
select(ShoppingList)
|
|
.where(ShoppingList.id == list_id)
|
|
.options(selectinload(ShoppingList.items).selectinload(ListItem.product))
|
|
)
|
|
result = await session.execute(stmt)
|
|
lst = result.scalar_one_or_none()
|
|
if not lst:
|
|
raise HTTPException(404, "Liste introuvable")
|
|
for field, value in payload.model_dump(exclude_unset=True).items():
|
|
setattr(lst, field, value)
|
|
await session.commit()
|
|
sorted_items = sorted(lst.items, key=lambda i: (i.sort_order or 999, str(i.id)))
|
|
return ShoppingListDetailResponse(
|
|
**_list_to_response(lst).model_dump(),
|
|
items=[_item_to_response(i) for i in sorted_items],
|
|
)
|
|
|
|
|
|
@router.delete("/lists/{list_id}", status_code=204)
|
|
async def delete_shopping_list(list_id: uuid.UUID, session: AsyncSession = Depends(get_session)):
|
|
lst = await session.get(ShoppingList, list_id)
|
|
if not lst:
|
|
raise HTTPException(404, "Liste introuvable")
|
|
await session.delete(lst)
|
|
await session.commit()
|
|
return Response(status_code=204)
|
|
|
|
|
|
# ── Items ─────────────────────────────────────────────────────────────────────
|
|
|
|
@router.post("/lists/{list_id}/items", response_model=ListItemResponse, status_code=201)
|
|
async def add_item(
|
|
list_id: uuid.UUID,
|
|
payload: ListItemCreate,
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
lst = await session.get(ShoppingList, list_id)
|
|
if not lst:
|
|
raise HTTPException(404, "Liste introuvable")
|
|
item = ListItem(list_id=list_id, **payload.model_dump())
|
|
session.add(item)
|
|
await session.commit()
|
|
stmt = (
|
|
select(ListItem)
|
|
.where(ListItem.id == item.id)
|
|
.options(selectinload(ListItem.product))
|
|
)
|
|
result = await session.execute(stmt)
|
|
item = result.scalar_one()
|
|
return _item_to_response(item)
|
|
|
|
|
|
@router.patch("/lists/{list_id}/items/{item_id}", response_model=ListItemResponse)
|
|
async def update_item(
|
|
list_id: uuid.UUID,
|
|
item_id: uuid.UUID,
|
|
payload: ListItemUpdate,
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
stmt = (
|
|
select(ListItem)
|
|
.where(ListItem.id == item_id, ListItem.list_id == list_id)
|
|
.options(selectinload(ListItem.product))
|
|
)
|
|
result = await session.execute(stmt)
|
|
item = result.scalar_one_or_none()
|
|
if not item:
|
|
raise HTTPException(404, "Article introuvable")
|
|
for field, value in payload.model_dump(exclude_unset=True).items():
|
|
setattr(item, field, value)
|
|
await session.commit()
|
|
await session.refresh(item, ["product"])
|
|
return _item_to_response(item)
|
|
|
|
|
|
@router.delete("/lists/{list_id}/items/{item_id}", status_code=204)
|
|
async def delete_item(
|
|
list_id: uuid.UUID,
|
|
item_id: uuid.UUID,
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
stmt = select(ListItem).where(ListItem.id == item_id, ListItem.list_id == list_id)
|
|
result = await session.execute(stmt)
|
|
item = result.scalar_one_or_none()
|
|
if not item:
|
|
raise HTTPException(404, "Article introuvable")
|
|
await session.delete(item)
|
|
await session.commit()
|
|
return Response(status_code=204)
|
|
|
|
|
|
@router.post("/lists/{list_id}/finish", response_model=ShoppingListDetailResponse)
|
|
async def finish_shopping(list_id: uuid.UUID, session: AsyncSession = Depends(get_session)):
|
|
stmt = (
|
|
select(ShoppingList)
|
|
.where(ShoppingList.id == list_id)
|
|
.options(selectinload(ShoppingList.items).selectinload(ListItem.product))
|
|
)
|
|
result = await session.execute(stmt)
|
|
lst = result.scalar_one_or_none()
|
|
if not lst:
|
|
raise HTTPException(404, "Liste introuvable")
|
|
|
|
lst.status = "done"
|
|
|
|
unchecked = [i for i in lst.items if not i.is_checked]
|
|
if unchecked:
|
|
new_list = ShoppingList(store_id=lst.store_id, status="draft")
|
|
session.add(new_list)
|
|
await session.flush()
|
|
for item in unchecked:
|
|
session.add(ListItem(
|
|
list_id=new_list.id,
|
|
product_id=item.product_id,
|
|
custom_name=item.custom_name,
|
|
quantity=item.quantity,
|
|
unit=item.unit,
|
|
sort_order=item.sort_order,
|
|
carried_over=True,
|
|
))
|
|
|
|
await session.commit()
|
|
sorted_items = sorted(lst.items, key=lambda i: (i.sort_order or 999, str(i.id)))
|
|
return ShoppingListDetailResponse(
|
|
**_list_to_response(lst).model_dump(),
|
|
items=[_item_to_response(i) for i in sorted_items],
|
|
)
|