72256f4fe9
Neuer Endpoint GET /items liefert alle Einzelstuecke ueber alle Produkte, angereichert mit Produkt-, Kategorie-, Lagerort- und Shop-Angaben (Beziehungen vorgeladen). ItemOut bekommt category_id/category_name. ProductOut bekommt image_version (Epoch der letzten Bildaenderung, None ohne Bild) - identisch zum ETag der Bild-Route, damit Clients Bilder cachen und nur bei Aenderung neu laden. Grundlage fuer die neue Einzelstueck-Liste (Web+iOS) und den iOS-Bild-Cache. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
"""Kleine gemeinsame Helfer für Router."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
|
||
from fastapi import HTTPException, status
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .models import Barcode, Category, CategoryTracking, Lot, Product, ProductImage
|
||
from .schemas import BarcodeOut, LocationMinStockOut, ProductOut
|
||
from .services.conversion import KIND_OF_BASE, display_unit_info
|
||
from .services.stock import current_stock
|
||
|
||
|
||
def product_tracking(db: Session, product: Product) -> str:
|
||
"""Verwaltungsart eines Artikels: aus seiner Kategorie abgeleitet.
|
||
|
||
Ohne Kategorie gilt "food" – so bleibt das Verhalten bestehender
|
||
(reiner Lebensmittel-)Installationen unverändert. Ein Artikel in einer
|
||
Gegenstands-Kategorie wird als "object" geführt.
|
||
"""
|
||
if product.category_id is None:
|
||
return CategoryTracking.food.value
|
||
cat = product.category or db.get(Category, product.category_id)
|
||
return cat.tracking if cat and cat.tracking else CategoryTracking.food.value
|
||
|
||
|
||
def product_to_out(db: Session, product: Product) -> ProductOut:
|
||
out = ProductOut.model_validate(product)
|
||
out.stock = current_stock(db, product.id)
|
||
out.expired_count = (
|
||
db.query(Lot)
|
||
.filter(
|
||
Lot.product_id == product.id,
|
||
Lot.best_before.isnot(None),
|
||
Lot.best_before < date.today(),
|
||
Lot.quantity > 0,
|
||
)
|
||
.count()
|
||
)
|
||
out.barcodes = [
|
||
BarcodeOut.model_validate(b)
|
||
for b in db.query(Barcode).filter(Barcode.product_id == product.id).order_by(Barcode.id).all()
|
||
]
|
||
out.category_name = product.category.name if product.category else None
|
||
# Bild-Version = Epoch der letzten Bildänderung (identisch zum ETag der
|
||
# Bild-Route), damit der Client nur geänderte Bilder neu lädt. None = kein Bild.
|
||
bild_ts = (
|
||
db.query(ProductImage.updated_at)
|
||
.filter(ProductImage.product_id == product.id)
|
||
.scalar()
|
||
)
|
||
out.image_version = int(bild_ts.timestamp()) if bild_ts is not None else None
|
||
out.tracking = CategoryTracking(product_tracking(db, product))
|
||
out.shop_name = product.shop.name if product.shop else None
|
||
out.kind = KIND_OF_BASE[product.base_unit].value
|
||
name, factor = display_unit_info(product)
|
||
out.unit_name = name
|
||
out.unit_factor = factor
|
||
|
||
# Mindestbestand in der Einheit anzeigen, in der er erfasst wurde.
|
||
if product.min_stock is not None:
|
||
if product.min_stock_in_packages and product.package_size:
|
||
out.min_stock_display = product.min_stock / product.package_size
|
||
out.min_stock_unit_label = "Pkg"
|
||
elif product.min_stock_unit is not None:
|
||
out.min_stock_display = product.min_stock / product.min_stock_unit.factor
|
||
out.min_stock_unit_label = product.min_stock_unit.name
|
||
else:
|
||
out.min_stock_display = product.min_stock / factor
|
||
out.min_stock_unit_label = name
|
||
|
||
out.location_min_stocks = [
|
||
LocationMinStockOut(
|
||
location_id=e.location_id,
|
||
location_name=e.location.name if e.location else None,
|
||
min_stock=e.min_stock,
|
||
)
|
||
for e in sorted(product.location_min_stocks, key=lambda x: x.id)
|
||
]
|
||
return out
|
||
|
||
|
||
def resolve_product(
|
||
db: Session, product_id: int | None, barcode: str | None
|
||
) -> Product:
|
||
"""Findet ein Produkt per ID oder Barcode; wirft 404, wenn keins passt."""
|
||
product: Product | None = None
|
||
if product_id is not None:
|
||
product = db.get(Product, product_id)
|
||
elif barcode:
|
||
product = db.query(Product).filter(Product.barcode == barcode).first()
|
||
else:
|
||
raise HTTPException(
|
||
status.HTTP_400_BAD_REQUEST, "product_id oder barcode erforderlich"
|
||
)
|
||
if product is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
|
||
return product
|