Scarriffle/Vorrania
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Verwaltbare Einheiten mit Umrechnung + Chargen bearbeiten + gezieltes Auslagern

Browse Source 
Einheiten (neu):
- Tabelle units (name, kind=count|weight|volume, factor, is_builtin); eingebaut
  Stueck/Gramm/Kilogramm/Milliliter/Liter, Admin kann eigene anlegen (z.B. Pfund=500g).
- Bestaende bleiben intern in kanonischer Basis (Stueck/Gramm/Milliliter);
  Product.display_unit_id und Group.min_stock_unit_id als nullable FKs.
- Neuer Umrechnungs-Service (services/conversion.py) ersetzt die feste Einheitenlogik;
  Ein-/Auslagern und Gruppen-Mindestbestand rechnen ueber den Faktor.
- Gruppen-Mindestbestand mit Einheit; Gruppenbestand summiert nur Produkte
  passender Art. Neue Verwaltungsseite "Einheiten" (Admin).
- Schonende Migration beim Start: ADD COLUMN IF NOT EXISTS (Postgres), damit
  bestehende Installationen ihre Daten behalten.

Chargen:
- PATCH /lots/{id} und DELETE /lots/{id}: Menge/MHD korrigieren, Charge loeschen
  (wird als Korrektur-Bewegung protokolliert). Bearbeitung im Produktdetail.

Auslagern:
- Optionales lot_id: gezielt aus einer bestimmten Charge/MHD abbuchen statt FEFO;
  Auswahl-Dropdown in der Auslagern-Seite.

Sonstiges: Roadmap aktualisiert, Tests fuer die Umrechnung.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-07-22 12:33:14 +02:00
parent d4e7ff8e3d
commit 0310cdfbd6
26 changed files with 1015 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
backend/app/crud.py
View File

@@ -9,6 +9,7 @@ from sqlalchemy.orm import Session
from .models import Lot, Product
from .schemas import ProductOut
from .services.conversion import KIND_OF_BASE, display_unit_info
from .services.stock import current_stock
@@ -25,6 +26,10 @@ def product_to_out(db: Session, product: Product) -> ProductOut:
)
.count()
)
out.kind = KIND_OF_BASE[product.base_unit].value
name, factor = display_unit_info(product)
out.unit_name = name
out.unit_factor = factor
return out

28
backend/app/main.py
View File

@@ -2,6 +2,7 @@ from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import text
from .config import get_settings
from .database import Base, SessionLocal, engine
@@ -12,20 +13,44 @@ from .routers import (
products,
settings as settings_router,
stock,
units,
users,
views,
)
from .seed import ensure_first_admin
from .seed import ensure_builtin_units, ensure_first_admin
settings = get_settings()
def _ensure_schema() -> None:
"""Schonende Migration: fehlende Spalten auf bestehenden Tabellen nachziehen.
create_all() legt nur fehlende Tabellen an, ändert aber keine bestehenden.
Auf Postgres holen wir neue nullable-Spalten per ADD COLUMN IF NOT EXISTS nach,
damit vorhandene Installationen ihre Daten behalten. Auf SQLite (Tests) sind
die Spalten bereits durch create_all vorhanden.
"""
if engine.dialect.name != "postgresql":
return
stmts = [
"ALTER TABLE products ADD COLUMN IF NOT EXISTS display_unit_id INTEGER "
"REFERENCES units(id) ON DELETE SET NULL",
"ALTER TABLE groups ADD COLUMN IF NOT EXISTS min_stock_unit_id INTEGER "
"REFERENCES units(id) ON DELETE SET NULL",
]
with engine.begin() as conn:
for stmt in stmts:
conn.execute(text(stmt))
@asynccontextmanager
async def lifespan(app: FastAPI):
# Tabellen anlegen (MVP: create_all statt Alembic-Migrationen).
Base.metadata.create_all(bind=engine)
_ensure_schema()
db = SessionLocal()
try:
ensure_builtin_units(db)
ensure_first_admin(db)
finally:
db.close()
@@ -57,5 +82,6 @@ app.include_router(products.router)
app.include_router(stock.router)
app.include_router(locations.router)
app.include_router(groups.router)
app.include_router(units.router)
app.include_router(views.router)
app.include_router(settings_router.router)

35
backend/app/models.py
View File

@@ -4,6 +4,7 @@ import enum
from datetime import date, datetime, timezone
from sqlalchemy import (
Boolean,
Date,
DateTime,
Enum,
@@ -40,6 +41,28 @@ class MovementType(str, enum.Enum):
adjust = "adjust"
class UnitKind(str, enum.Enum):
"""Art einer Einheit. Bestimmt die kanonische Basiseinheit für die Speicherung."""
count = "count" # Basis: Stück
weight = "weight" # Basis: Gramm
volume = "volume" # Basis: Milliliter
class Unit(Base):
"""Vom Admin verwaltbare Einheit mit Umrechnungsfaktor zur kanonischen Basis.
factor = wie viele Basiseinheiten 1 dieser Einheit entsprechen
(z.B. Kilogramm: kind=weight, factor=1000; Liter: kind=volume, factor=1000).
"""
__tablename__ = "units"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
kind: Mapped[UnitKind] = mapped_column(Enum(UnitKind), nullable=False)
factor: Mapped[float] = mapped_column(Float, nullable=False, default=1.0)
is_builtin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
class User(Base):
__tablename__ = "users"
@@ -55,10 +78,14 @@ class Group(Base):
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(120), unique=True, nullable=False)
# Group-level minimum stock is a Schritt-3 feature; column kept for forward-compat.
min_stock: Mapped[float | None] = mapped_column(Float, nullable=True)
# Einheit des Gruppen-Mindestbestands (z.B. Kilogramm). NULL = zählt in Basiseinheiten.
min_stock_unit_id: Mapped[int | None] = mapped_column(
ForeignKey("units.id", ondelete="SET NULL"), nullable=True
)
products: Mapped[list[Product]] = relationship(back_populates="group")
min_stock_unit: Mapped[Unit | None] = relationship()
class Location(Base):
@@ -82,9 +109,14 @@ class Product(Base):
brand: Mapped[str | None] = mapped_column(String(255), nullable=True)
image_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
# Kanonische Basiseinheit (piece/gram/milliliter): bestimmt Speicherung + Art (kind).
base_unit: Mapped[BaseUnit] = mapped_column(
Enum(BaseUnit), nullable=False, default=BaseUnit.piece
)
# Bevorzugte Anzeige-/Eingabeeinheit (z.B. Kilogramm). NULL = Basiseinheit selbst.
display_unit_id: Mapped[int | None] = mapped_column(
ForeignKey("units.id", ondelete="SET NULL"), nullable=True
)
# Number of base units contained in one package (e.g. 500 g per package).
package_size: Mapped[float | None] = mapped_column(Float, nullable=True)
@@ -98,6 +130,7 @@ class Product(Base):
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
group: Mapped[Group | None] = relationship(back_populates="products")
display_unit: Mapped[Unit | None] = relationship()
lots: Mapped[list[Lot]] = relationship(
back_populates="product", cascade="all, delete-orphan"
)

19
backend/app/routers/groups.py
View File

@@ -5,6 +5,7 @@ from ..database import get_db
from ..deps import get_current_user, require_admin
from ..models import Group, Product, User
from ..schemas import GroupCreate, GroupOut, GroupUpdate
from ..services.conversion import BASE_OF_KIND
from ..services.stock import current_stock
router = APIRouter(prefix="/groups", tags=["groups"])
@@ -14,7 +15,17 @@ def _group_to_out(db: Session, group: Group) -> GroupOut:
out = GroupOut.model_validate(group)
products = db.query(Product).filter(Product.group_id == group.id).all()
out.product_count = len(products)
out.stock = float(sum(current_stock(db, p.id) for p in products))
unit = group.min_stock_unit
if unit is not None:
# Bestand nur über Produkte gleicher Art, in der Gruppen-Einheit ausgedrückt.
base = BASE_OF_KIND[unit.kind]
matching = [p for p in products if p.base_unit == base]
stock_base = float(sum(current_stock(db, p.id) for p in matching))
out.stock = stock_base / unit.factor
out.min_stock_unit_name = unit.name
out.kind = unit.kind.value
else:
out.stock = float(sum(current_stock(db, p.id) for p in products))
return out
@@ -34,7 +45,11 @@ def create_group(
) -> GroupOut:
if db.query(Group).filter(Group.name == payload.name).first():
raise HTTPException(status.HTTP_409_CONFLICT, "Gruppe existiert bereits")
group = Group(name=payload.name, min_stock=payload.min_stock)
group = Group(
name=payload.name,
min_stock=payload.min_stock,
min_stock_unit_id=payload.min_stock_unit_id,
)
db.add(group)
db.commit()
db.refresh(group)

22
backend/app/routers/products.py
View File

@@ -7,6 +7,7 @@ from ..deps import get_current_user, require_admin
from ..models import BaseUnit, Product, User
from ..off import lookup_barcode
from ..schemas import LookupResult, ProductCreate, ProductOut, ProductUpdate
from ..services.conversion import ConversionError, resolve_product_unit
router = APIRouter(prefix="/products", tags=["products"])
@@ -66,12 +67,20 @@ def create_product(
raise HTTPException(
status.HTTP_409_CONFLICT, "Ein Produkt mit diesem Barcode existiert bereits"
)
base_unit = payload.base_unit
display_unit_id = None
if payload.unit_id is not None:
try:
base_unit, display_unit_id = resolve_product_unit(db, payload.unit_id)
except ConversionError as exc:
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
product = Product(
barcode=payload.barcode or None,
name=payload.name,
brand=payload.brand,
image_url=payload.image_url,
base_unit=payload.base_unit,
base_unit=base_unit,
display_unit_id=display_unit_id,
package_size=payload.package_size,
group_id=payload.group_id,
min_stock=payload.min_stock,
@@ -105,6 +114,17 @@ def update_product(
raise HTTPException(
status.HTTP_409_CONFLICT, "Ein anderes Produkt hat diesen Barcode bereits"
)
# Einheit: unit_id (falls gesetzt) bestimmt base_unit + Anzeigeeinheit.
if "unit_id" in data:
unit_id = data.pop("unit_id")
data.pop("base_unit", None) # unit_id hat Vorrang
if unit_id is None:
product.display_unit_id = None
else:
try:
product.base_unit, product.display_unit_id = resolve_product_unit(db, unit_id)
except ConversionError as exc:
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
for field, value in data.items():
setattr(product, field, value)
db.commit()

111
backend/app/routers/stock.py
View File

@@ -4,7 +4,7 @@ from sqlalchemy.orm import Session
from ..crud import resolve_product
from ..database import get_db
from ..deps import get_current_user
from ..models import Lot, User
from ..models import Lot, Movement, MovementType, User
from ..schemas import (
BatchCheckInRequest,
BatchCheckInResponse,
@@ -13,9 +13,16 @@ from ..schemas import (
CheckOutRequest,
CheckOutResponse,
LotOut,
LotUpdate,
)
from ..services.conversion import ConversionError
from ..services.stock import (
StockError,
check_in,
check_out,
check_out_lot,
current_stock,
)
from ..services.stock import StockError, check_in, check_out, current_stock
from ..services.units import UnitError
router = APIRouter(tags=["stock"])
@@ -38,7 +45,7 @@ def stock_checkin(
user=user,
note=payload.note,
)
except UnitError as exc:
except ConversionError as exc:
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
db.commit()
db.refresh(lot)
@@ -74,7 +81,7 @@ def stock_checkin_batch(
note=payload.note,
)
)
except UnitError as exc:
except ConversionError as exc:
db.rollback()
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
db.commit()
@@ -94,15 +101,31 @@ def stock_checkout(
) -> CheckOutResponse:
product = resolve_product(db, payload.product_id, payload.barcode)
try:
affected = check_out(
db,
product=product,
quantity=payload.quantity,
unit=payload.unit,
user=user,
note=payload.note,
)
except UnitError as exc:
if payload.lot_id is not None:
lot = db.get(Lot, payload.lot_id)
if lot is None or lot.product_id != product.id:
raise HTTPException(
status.HTTP_404_NOT_FOUND, "Charge gehört nicht zu diesem Produkt"
)
affected = check_out_lot(
db,
product=product,
lot=lot,
quantity=payload.quantity,
unit=payload.unit,
user=user,
note=payload.note,
)
else:
affected = check_out(
db,
product=product,
quantity=payload.quantity,
unit=payload.unit,
user=user,
note=payload.note,
)
except ConversionError as exc:
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
except StockError as exc:
raise HTTPException(status.HTTP_409_CONFLICT, str(exc)) from exc
@@ -122,3 +145,63 @@ def list_lots(
if product_id is not None:
query = query.filter(Lot.product_id == product_id)
return query.order_by(Lot.best_before.is_(None), Lot.best_before).all()
@router.patch("/lots/{lot_id}", response_model=LotOut)
def update_lot(
lot_id: int,
payload: LotUpdate,
db: Session = Depends(get_db),
user: User = Depends(get_current_user),
) -> Lot:
"""Korrigiert eine Charge (Menge in Basiseinheiten, MHD, Lagerort)."""
lot = db.get(Lot, lot_id)
if lot is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Charge nicht gefunden")
data = payload.model_dump(exclude_unset=True)
old_quantity = lot.quantity
for field, value in data.items():
setattr(lot, field, value)
delta = lot.quantity - old_quantity
if abs(delta) > 1e-9:
db.add(
Movement(
product_id=lot.product_id,
lot_id=lot.id,
user_id=user.id,
type=MovementType.adjust,
quantity=delta,
unit_used="base",
note="Charge korrigiert",
)
)
db.commit()
db.refresh(lot)
return lot
@router.delete("/lots/{lot_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_lot(
lot_id: int,
db: Session = Depends(get_db),
user: User = Depends(get_current_user),
) -> None:
"""Entfernt eine Charge komplett aus dem Bestand."""
lot = db.get(Lot, lot_id)
if lot is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Charge nicht gefunden")
db.add(
Movement(
product_id=lot.product_id,
lot_id=None,
user_id=user.id,
type=MovementType.adjust,
quantity=-lot.quantity,
unit_used="base",
note="Charge gelöscht",
)
)
db.delete(lot)
db.commit()

56
backend/app/routers/units.py Normal file
View File

@@ -0,0 +1,56 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import func
from sqlalchemy.orm import Session
from ..database import get_db
from ..deps import get_current_user, require_admin
from ..models import Group, Product, Unit, User
from ..schemas import UnitCreate, UnitOut
router = APIRouter(prefix="/units", tags=["units"])
@router.get("", response_model=list[UnitOut])
def list_units(
db: Session = Depends(get_db), _: User = Depends(get_current_user)
) -> list[Unit]:
return db.query(Unit).order_by(Unit.kind, Unit.factor).all()
@router.post("", response_model=UnitOut, status_code=status.HTTP_201_CREATED)
def create_unit(
payload: UnitCreate,
db: Session = Depends(get_db),
_: User = Depends(require_admin),
) -> Unit:
exists = db.query(Unit).filter(func.lower(Unit.name) == payload.name.lower()).first()
if exists:
raise HTTPException(status.HTTP_409_CONFLICT, "Einheit existiert bereits")
unit = Unit(name=payload.name, kind=payload.kind, factor=payload.factor, is_builtin=False)
db.add(unit)
db.commit()
db.refresh(unit)
return unit
@router.delete("/{unit_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_unit(
unit_id: int,
db: Session = Depends(get_db),
_: User = Depends(require_admin),
) -> None:
unit = db.get(Unit, unit_id)
if unit is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Einheit nicht gefunden")
if unit.is_builtin:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Eingebaute Einheiten können nicht gelöscht werden")
in_use = (
db.query(Product).filter(Product.display_unit_id == unit_id).first()
or db.query(Group).filter(Group.min_stock_unit_id == unit_id).first()
)
if in_use:
raise HTTPException(
status.HTTP_409_CONFLICT, "Einheit wird noch von Produkten/Gruppen verwendet"
)
db.delete(unit)
db.commit()

14
backend/app/routers/views.py
View File

@@ -7,6 +7,7 @@ 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
from ..services.stock import current_stock
from .settings import get_expiry_warning_days
@@ -56,8 +57,16 @@ def group_shopping_list(
db.query(Group).filter(Group.min_stock.isnot(None), Group.min_stock > 0).all()
)
for group in groups:
products = db.query(Product).filter(Product.group_id == group.id).all()
stock = float(sum(current_stock(db, p.id) for p in products))
unit = group.min_stock_unit
if unit is not None:
base = BASE_OF_KIND[unit.kind]
products = [p for p in group.products if p.base_unit == base]
stock = float(sum(current_stock(db, p.id) for p in products)) / unit.factor
unit_name = unit.name
else:
products = list(group.products)
stock = float(sum(current_stock(db, p.id) for p in products))
unit_name = ""
if stock < group.min_stock:
items.append(
GroupShoppingItem(
@@ -66,6 +75,7 @@ def group_shopping_list(
stock=stock,
min_stock=group.min_stock,
deficit=group.min_stock - stock,
unit_name=unit_name,
product_count=len(products),
)
)

42
backend/app/schemas.py
View File

@@ -4,7 +4,23 @@ from datetime import date, datetime
from pydantic import BaseModel, ConfigDict, Field
from .models import BaseUnit, Role
from .models import BaseUnit, Role, UnitKind
# ---- Units ----
class UnitOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
kind: UnitKind
factor: float
is_builtin: bool
class UnitCreate(BaseModel):
name: str = Field(min_length=1, max_length=64)
kind: UnitKind
factor: float = Field(gt=0)
# ---- Auth / Users ----
@@ -40,19 +56,24 @@ class GroupOut(BaseModel):
id: int
name: str
min_stock: float | None = None
min_stock_unit_id: int | None = None
# angereichert:
product_count: int = 0
stock: float = 0.0
stock: float = 0.0 # Bestand in Basiseinheiten
min_stock_unit_name: str | None = None
kind: str | None = None # Art der Mindestbestand-Einheit
class GroupCreate(BaseModel):
name: str = Field(min_length=1, max_length=120)
min_stock: float | None = Field(default=None, ge=0)
min_stock_unit_id: int | None = None
class GroupUpdate(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=120)
min_stock: float | None = Field(default=None, ge=0)
min_stock_unit_id: int | None = None
# ---- Locations ----
@@ -75,6 +96,8 @@ class ProductBase(BaseModel):
brand: str | None = None
image_url: str | None = None
base_unit: BaseUnit = BaseUnit.piece
# Optionale verwaltete Einheit; setzt base_unit anhand ihrer Art und die Anzeigeeinheit.
unit_id: int | None = None
package_size: float | None = Field(default=None, gt=0)
group_id: int | None = None
min_stock: float | None = Field(default=None, ge=0)
@@ -90,6 +113,7 @@ class ProductUpdate(BaseModel):
brand: str | None = None
image_url: str | None = None
base_unit: BaseUnit | None = None
unit_id: int | None = None
package_size: float | None = Field(default=None, gt=0)
group_id: int | None = None
min_stock: float | None = Field(default=None, ge=0)
@@ -103,6 +127,7 @@ class ProductOut(BaseModel):
brand: str | None
image_url: str | None
base_unit: BaseUnit
display_unit_id: int | None = None
package_size: float | None
group_id: int | None
min_stock: float | None
@@ -111,6 +136,9 @@ class ProductOut(BaseModel):
# angereichert:
stock: float = 0.0
expired_count: int = 0
kind: str = ""
unit_name: str = ""
unit_factor: float = 1.0
class LookupResult(BaseModel):
@@ -135,6 +163,8 @@ class CheckOutRequest(BaseModel):
barcode: str | None = None
quantity: float = Field(gt=0)
unit: str
# Optional: gezielt aus dieser Charge abbuchen (sonst automatisch FEFO).
lot_id: int | None = None
note: str | None = None
@@ -148,6 +178,13 @@ class LotOut(BaseModel):
created_at: datetime
class LotUpdate(BaseModel):
"""Korrektur einer Charge (Vertipper beim Einlagern o.ä.)."""
quantity: float | None = Field(default=None, gt=0)
best_before: date | None = None
location_id: int | None = None
class CheckInResponse(BaseModel):
lot: LotOut
product_stock: float
@@ -205,6 +242,7 @@ class GroupShoppingItem(BaseModel):
stock: float
min_stock: float
deficit: float
unit_name: str = ""
product_count: int

23
backend/app/seed.py
View File

@@ -1,11 +1,30 @@
"""Erstellt beim ersten Start einen Admin-Benutzer, falls noch keiner existiert."""
"""Startup-Seeds: erster Admin-Benutzer und die eingebauten Einheiten."""
from sqlalchemy.orm import Session
from .config import get_settings
from .models import Role, User
from .models import Role, Unit, UnitKind, User
from .security import hash_password
# (Name, Art, Faktor zur kanonischen Basiseinheit)
BUILTIN_UNITS: list[tuple[str, UnitKind, float]] = [
("Stück", UnitKind.count, 1.0),
("Gramm", UnitKind.weight, 1.0),
("Kilogramm", UnitKind.weight, 1000.0),
("Milliliter", UnitKind.volume, 1.0),
("Liter", UnitKind.volume, 1000.0),
]
def ensure_builtin_units(db: Session) -> None:
changed = False
for name, kind, factor in BUILTIN_UNITS:
if not db.query(Unit).filter(Unit.name == name).first():
db.add(Unit(name=name, kind=kind, factor=factor, is_builtin=True))
changed = True
if changed:
db.commit()
def ensure_first_admin(db: Session) -> None:
settings = get_settings()

90
backend/app/services/conversion.py Normal file
View File

@@ -0,0 +1,90 @@
"""Einheiten-Umrechnung auf Basis der verwaltbaren Unit-Tabelle.
Bestände werden immer in der kanonischen Basiseinheit der jeweiligen Art
gespeichert: Stück (count), Gramm (weight), Milliliter (volume). Verwaltete
Einheiten (z.B. Kilogramm, Liter, Pfund) rechnen über ihren Faktor dorthin um.
"""
from __future__ import annotations
from sqlalchemy import func
from sqlalchemy.orm import Session
from ..models import BaseUnit, Product, Unit, UnitKind
BASE_OF_KIND: dict[UnitKind, BaseUnit] = {
UnitKind.count: BaseUnit.piece,
UnitKind.weight: BaseUnit.gram,
UnitKind.volume: BaseUnit.milliliter,
}
KIND_OF_BASE: dict[BaseUnit, UnitKind] = {v: k for k, v in BASE_OF_KIND.items()}
BASE_LABEL: dict[BaseUnit, str] = {
BaseUnit.piece: "Stück",
BaseUnit.gram: "Gramm",
BaseUnit.milliliter: "Milliliter",
}
PACKAGE_TOKENS = {"package", "packung", "pkg", "pack"}
# Kürzel/Aliase, die direkt auf eingebaute Einheitennamen zeigen.
_ALIAS = {
"g": "gramm", "gram": "gramm", "gramm": "gramm",
"kg": "kilogramm", "kilogramm": "kilogramm",
"ml": "milliliter", "milliliter": "milliliter",
"l": "liter", "liter": "liter",
"stück": "stück", "stueck": "stück", "st": "stück", "stk": "stück", "piece": "stück",
}
class ConversionError(ValueError):
"""Fachlicher Fehler bei der Einheiten-Umrechnung."""
def kind_of_product(product: Product) -> UnitKind:
return KIND_OF_BASE[product.base_unit]
def find_unit(db: Session, token: str) -> Unit | None:
t = token.strip().lower()
t = _ALIAS.get(t, t)
return db.query(Unit).filter(func.lower(Unit.name) == t).first()
def to_base(db: Session, product: Product, quantity: float, unit_token: str) -> float:
"""Rechnet eine Menge (in unit_token) in die Basiseinheit des Produkts um."""
if quantity <= 0:
raise ConversionError("Menge muss größer als 0 sein")
t = (unit_token or "").strip().lower()
if t in PACKAGE_TOKENS:
if not product.package_size or product.package_size <= 0:
raise ConversionError(
"Für dieses Produkt ist keine Packungsgröße hinterlegt."
)
return quantity * product.package_size
unit = find_unit(db, t)
if unit is None:
raise ConversionError(f"Unbekannte Einheit: {unit_token}")
if unit.kind != kind_of_product(product):
raise ConversionError(
f"Einheit '{unit.name}' passt nicht zur Art des Produkts "
f"({BASE_LABEL[product.base_unit]})."
)
return quantity * unit.factor
def resolve_product_unit(db: Session, unit_id: int) -> tuple[BaseUnit, int]:
"""Ermittelt zu einer gewählten Einheit die kanonische Basiseinheit + Anzeige-ID."""
unit = db.get(Unit, unit_id)
if unit is None:
raise ConversionError("Einheit nicht gefunden")
return BASE_OF_KIND[unit.kind], unit.id
def display_unit_info(product: Product) -> tuple[str, float]:
"""Name und Faktor der Anzeigeeinheit des Produkts (Fallback: Basiseinheit)."""
if product.display_unit is not None:
return product.display_unit.name, product.display_unit.factor
return BASE_LABEL[product.base_unit], 1.0

41
backend/app/services/stock.py
View File

@@ -8,7 +8,7 @@ from sqlalchemy import asc
from sqlalchemy.orm import Session
from ..models import Lot, Movement, MovementType, Product, User
from .units import to_base_quantity
from .conversion import to_base
class StockError(ValueError):
@@ -34,7 +34,7 @@ def check_in(
note: str | None = None,
) -> Lot:
"""Legt eine neue Charge an und protokolliert die Bewegung."""
quantity_base = to_base_quantity(product, quantity, unit)
quantity_base = to_base(db, product, quantity, unit)
lot = Lot(
product_id=product.id,
@@ -59,6 +59,41 @@ def check_in(
return lot
def check_out_lot(
db: Session,
product: Product,
lot: Lot,
quantity: float,
unit: str,
user: User | None,
note: str | None = None,
) -> list[dict]:
"""Bucht gezielt von EINER Charge ab (manuelle Auswahl statt FEFO)."""
needed = to_base(db, product, quantity, unit)
if needed > lot.quantity + 1e-9:
raise StockError(
f"Diese Charge hat nur {lot.quantity:g} {product.base_unit.value} "
f"(benötigt {needed:g})."
)
lot.quantity -= needed
db.add(
Movement(
product_id=product.id,
lot_id=lot.id,
user_id=user.id if user else None,
type=MovementType.out,
quantity=needed,
unit_used=unit,
note=note,
)
)
affected = [{"lot_id": lot.id, "quantity": needed}]
if lot.quantity <= 1e-9:
db.delete(lot)
return affected
def _fefo_lots(db: Session, product_id: int) -> list[Lot]:
"""Lots eines Produkts, sortiert nach Ablaufdatum (NULL zuletzt), dann Alter."""
lots = (
@@ -88,7 +123,7 @@ def check_out(
Gibt die Liste der betroffenen Chargen mit abgebuchter Menge zurück.
Wirft StockError, wenn der Gesamtbestand nicht ausreicht.
"""
needed = to_base_quantity(product, quantity, unit)
needed = to_base(db, product, quantity, unit)
available = current_stock(db, product.id)
if needed > available + 1e-9:
raise StockError(

2
backend/tests/conftest.py
View File

@@ -5,6 +5,7 @@ from sqlalchemy.pool import StaticPool
from app.database import Base
from app.models import BaseUnit, Product
from app.seed import ensure_builtin_units
@pytest.fixture()
@@ -17,6 +18,7 @@ def db():
Base.metadata.create_all(bind=engine)
TestingSession = sessionmaker(bind=engine, autoflush=False, autocommit=False)
session = TestingSession()
ensure_builtin_units(session) # Einheiten (Gramm, Kilogramm, …) für Umrechnung
try:
yield session
finally:

43
backend/tests/test_conversion.py Normal file
View File

@@ -0,0 +1,43 @@
import pytest
from app.models import BaseUnit, Product
from app.services.conversion import ConversionError, to_base
def test_base_unit(db, rice):
assert to_base(db, rice, 200, "Gramm") == 200
assert to_base(db, rice, 200, "g") == 200 # Kürzel-Alias
def test_kilogram_converts_to_grams(db, rice):
assert to_base(db, rice, 1, "Kilogramm") == 1000
assert to_base(db, rice, 2.5, "kg") == 2500
def test_package_uses_package_size(db, rice):
# rice: package_size = 500 g
assert to_base(db, rice, 3, "package") == 1500
def test_wrong_kind_rejected(db, rice):
# rice ist ein Gewichts-Produkt -> Volumeneinheit passt nicht
with pytest.raises(ConversionError):
to_base(db, rice, 1, "Liter")
def test_unknown_unit_rejected(db, rice):
with pytest.raises(ConversionError):
to_base(db, rice, 1, "Eimer")
def test_volume_product(db):
milk = Product(name="Milch", base_unit=BaseUnit.milliliter, package_size=1000)
db.add(milk)
db.commit()
assert to_base(db, milk, 1, "Liter") == 1000
assert to_base(db, milk, 500, "ml") == 500
def test_non_positive_quantity_rejected(db, rice):
with pytest.raises(ConversionError):
to_base(db, rice, 0, "Gramm")