Backend (FastAPI + PostgreSQL): Chargen mit MHD, FEFO-Auslagern, Einheiten-Umrechnung (Stueck/g/ml + Packungen), Open-Food-Facts-Lookup mit lokalem Fallback, JWT-Auth mit Rollen (Admin/Nutzer), erster Admin beim Setup, Einkaufsliste, Ablaufwarnung, Lagerorte, pytest fuer FEFO. Web-UI (React/Vite): Login, Dashboard, Ein-/Auslagern, Produkte, Lagerorte, Benutzerverwaltung, Einkaufsliste - rollenabhaengig. Deploy: docker-compose + install.sh (Docker-Autoinstall, Secrets), README und Roadmap fuer Schritt 2 (iOS) und Schritt 3. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
from datetime import date, timedelta
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..database import get_db
|
|
from ..deps import get_current_user
|
|
from ..models import Lot, Product, User
|
|
from ..schemas import ExpiringItem, ShoppingItem
|
|
from ..services.stock import current_stock
|
|
from .settings import get_expiry_warning_days
|
|
|
|
router = APIRouter(tags=["views"])
|
|
|
|
|
|
@router.get("/shopping-list", response_model=list[ShoppingItem])
|
|
def shopping_list(
|
|
db: Session = Depends(get_db), _: User = Depends(get_current_user)
|
|
) -> list[ShoppingItem]:
|
|
"""Produkte, deren Bestand unter dem Mindestbestand liegt."""
|
|
items: list[ShoppingItem] = []
|
|
products = (
|
|
db.query(Product)
|
|
.filter(Product.min_stock.isnot(None), Product.min_stock > 0)
|
|
.all()
|
|
)
|
|
for product in products:
|
|
stock = current_stock(db, product.id)
|
|
if stock < product.min_stock:
|
|
items.append(
|
|
ShoppingItem(
|
|
product_id=product.id,
|
|
name=product.name,
|
|
base_unit=product.base_unit,
|
|
stock=stock,
|
|
min_stock=product.min_stock,
|
|
deficit=product.min_stock - stock,
|
|
)
|
|
)
|
|
items.sort(key=lambda i: i.deficit, reverse=True)
|
|
return items
|
|
|
|
|
|
@router.get("/expiring", response_model=list[ExpiringItem])
|
|
def expiring(
|
|
days: int | None = None,
|
|
db: Session = Depends(get_db),
|
|
_: User = Depends(get_current_user),
|
|
) -> list[ExpiringItem]:
|
|
"""Chargen, die innerhalb der Warnfrist ablaufen (oder schon abgelaufen sind)."""
|
|
if days is None:
|
|
days = get_expiry_warning_days(db)
|
|
today = date.today()
|
|
threshold = today + timedelta(days=days)
|
|
|
|
lots = (
|
|
db.query(Lot)
|
|
.filter(Lot.best_before.isnot(None), Lot.best_before <= threshold, Lot.quantity > 0)
|
|
.order_by(Lot.best_before)
|
|
.all()
|
|
)
|
|
result: list[ExpiringItem] = []
|
|
for lot in lots:
|
|
product = lot.product
|
|
result.append(
|
|
ExpiringItem(
|
|
lot_id=lot.id,
|
|
product_id=product.id,
|
|
product_name=product.name,
|
|
quantity=lot.quantity,
|
|
base_unit=product.base_unit,
|
|
best_before=lot.best_before,
|
|
days_left=(lot.best_before - today).days,
|
|
)
|
|
)
|
|
return result
|