Backend: Mindestbestand je Lagerort für Produkte und Gruppen
Zusaetzlich zum globalen Mindestbestand: je Produkt und je Gruppe laesst sich pro
Lagerort ein Mindestbestand (in Artikeleinheiten) hinterlegen.
- Neue Tabellen product_location_min_stock / group_location_min_stock.
- PUT /products/{id}/location-min-stock und /groups/{id}/location-min-stock
ersetzen die Eintraege; Produkt-/Gruppen-Ausgabe liefert sie mit.
- Neue Einkaufsliste GET /shopping-list/by-location: Bedarfe je Ort (Produkte +
Gruppen), Bestand-am-Ort gegen Mindestbestand-am-Ort.
- Helfer location_stock_base (Bestand je Ort). 4 neue Tests, Suite 144 gruen.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
from collections import defaultdict
|
||||
from datetime import date, timedelta
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
@@ -5,10 +6,27 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from ..database import get_db
|
||||
from ..deps import get_current_user
|
||||
from ..models import Group, Lot, Movement, Product, User
|
||||
from ..schemas import ExpiringItem, GroupShoppingItem, MovementOut, ShoppingItem
|
||||
from ..services.conversion import BASE_OF_KIND, display_unit_info
|
||||
from ..services.stock import current_stock
|
||||
from ..models import (
|
||||
Group,
|
||||
GroupLocationMinStock,
|
||||
Location,
|
||||
Lot,
|
||||
Movement,
|
||||
Product,
|
||||
ProductLocationMinStock,
|
||||
User,
|
||||
)
|
||||
from ..schemas import (
|
||||
ExpiringItem,
|
||||
GroupShoppingItem,
|
||||
LocationNeedGroup,
|
||||
LocationNeedProduct,
|
||||
LocationNeeds,
|
||||
MovementOut,
|
||||
ShoppingItem,
|
||||
)
|
||||
from ..services.conversion import BASE_OF_KIND, article_unit, display_unit_info
|
||||
from ..services.stock import current_stock, location_stock_base
|
||||
from .settings import get_expiry_warning_days
|
||||
|
||||
router = APIRouter(tags=["views"])
|
||||
@@ -83,6 +101,64 @@ def group_shopping_list(
|
||||
return items
|
||||
|
||||
|
||||
@router.get("/shopping-list/by-location", response_model=list[LocationNeeds])
|
||||
def shopping_list_by_location(
|
||||
db: Session = Depends(get_db), _: User = Depends(get_current_user)
|
||||
) -> list[LocationNeeds]:
|
||||
"""Bedarfe je Lagerort: Produkte und Gruppen, deren Bestand AN DIESEM ORT
|
||||
unter dem dort hinterlegten Mindestbestand liegt."""
|
||||
prod_needs: dict[int, list[LocationNeedProduct]] = defaultdict(list)
|
||||
group_needs: dict[int, list[LocationNeedGroup]] = defaultdict(list)
|
||||
|
||||
for e in db.query(ProductLocationMinStock).all():
|
||||
product = db.get(Product, e.product_id)
|
||||
if product is None:
|
||||
continue
|
||||
faktor, label = article_unit(product)
|
||||
stock = location_stock_base(db, product, e.location_id) / (faktor or 1.0)
|
||||
if stock < e.min_stock:
|
||||
prod_needs[e.location_id].append(LocationNeedProduct(
|
||||
product_id=product.id, name=product.name, unit_label=label,
|
||||
stock=round(stock, 3), min_stock=e.min_stock,
|
||||
deficit=round(e.min_stock - stock, 3),
|
||||
))
|
||||
|
||||
for e in db.query(GroupLocationMinStock).all():
|
||||
group = db.get(Group, e.group_id)
|
||||
if group is None:
|
||||
continue
|
||||
unit = group.min_stock_unit
|
||||
if unit is not None:
|
||||
base = BASE_OF_KIND[unit.kind]
|
||||
matching = [p for p in group.products if p.base_unit == base]
|
||||
stock = sum(location_stock_base(db, p, e.location_id) for p in matching) / unit.factor
|
||||
unit_name = unit.name
|
||||
else:
|
||||
stock = float(sum(location_stock_base(db, p, e.location_id) for p in group.products))
|
||||
unit_name = ""
|
||||
if stock < e.min_stock:
|
||||
group_needs[e.location_id].append(LocationNeedGroup(
|
||||
group_id=group.id, name=group.name, unit_name=unit_name,
|
||||
stock=round(stock, 3), min_stock=e.min_stock,
|
||||
deficit=round(e.min_stock - stock, 3),
|
||||
))
|
||||
|
||||
loc_ids = set(prod_needs) | set(group_needs)
|
||||
namen = {
|
||||
loc.id: loc.name
|
||||
for loc in db.query(Location).filter(Location.id.in_(loc_ids)).all()
|
||||
}
|
||||
result: list[LocationNeeds] = []
|
||||
for loc_id in sorted(loc_ids, key=lambda i: namen.get(i, "")):
|
||||
result.append(LocationNeeds(
|
||||
location_id=loc_id,
|
||||
location_name=namen.get(loc_id, "?"),
|
||||
products=sorted(prod_needs.get(loc_id, []), key=lambda x: x.deficit, reverse=True),
|
||||
groups=sorted(group_needs.get(loc_id, []), key=lambda x: x.deficit, reverse=True),
|
||||
))
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/expiring", response_model=list[ExpiringItem])
|
||||
def expiring(
|
||||
days: int | None = None,
|
||||
|
||||
Reference in New Issue
Block a user