Backend: Einzelstücke (Items) mit UID je Gegenstand

Gegenstands-Produkte koennen als Einzelstuecke gefuehrt werden (Product.individual):
jedes physische Stueck ist ein Item mit eigener kurzer UID (fuer QR), eigenem
Lagerort, Kaufdatum, Garantie, Bezugsquelle und Notiz. So laesst sich dasselbe
Modell mehrfach getrennt fuehren (Powerbank 2024 + 2025).

Neu: models.Item + Product.individual (+Migration), Schemas, services/items.py
(UID-Erzeugung, Anreicherung), routers/items.py (CRUD, /items/by-uid/{uid} zur
QR-Aufloesung, /items/{id}/remove mit Grund als Bewegung). current_stock zaehlt
bei Einzelstueck-Produkten die Items. Tests + HTTP-Smoke gruen.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-07-26 00:26:15 +02:00
parent 68e27c1868
commit 5ce4411d1e
8 changed files with 326 additions and 2 deletions

View File

@@ -0,0 +1,158 @@
"""Einzelstücke (Items) eines Gegenstands: anlegen, ändern, entfernen, per UID finden.
Ein Item ist ein physisches Exemplar mit eigener UID/QR und eigenen Angaben
(Lagerort, Kaufdatum, Garantie, Bezugsquelle, Notiz). Nur sinnvoll für
Gegenstands-Produkte mit ``Product.individual``.
"""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from ..database import get_db
from ..deps import get_current_user, require_admin
from ..models import Item, Location, Movement, MovementType, Product, Shop, User
from ..schemas import ItemCreate, ItemOut, ItemRemove, ItemUpdate
from ..services.items import generate_uid, item_to_out
router = APIRouter(tags=["items"])
def _product_or_404(db: Session, product_id: int) -> Product:
product = db.get(Product, product_id)
if product is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Produkt nicht gefunden")
return product
def _item_or_404(db: Session, item_id: int) -> Item:
item = db.get(Item, item_id)
if item is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Einzelstück nicht gefunden")
return item
def _check_refs(db: Session, shop_id: int | None, location_id: int | None) -> None:
if shop_id is not None and db.get(Shop, shop_id) is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Shop nicht gefunden")
if location_id is not None and db.get(Location, location_id) is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Lagerort nicht gefunden")
@router.get("/products/{product_id}/items", response_model=list[ItemOut])
def list_items(
product_id: int,
db: Session = Depends(get_db),
_: User = Depends(get_current_user),
) -> list[ItemOut]:
_product_or_404(db, product_id)
rows = db.query(Item).filter(Item.product_id == product_id).order_by(Item.id).all()
return [item_to_out(i) for i in rows]
@router.post(
"/products/{product_id}/items", response_model=list[ItemOut],
status_code=status.HTTP_201_CREATED,
)
def create_items(
product_id: int,
payload: ItemCreate,
db: Session = Depends(get_db),
_: User = Depends(require_admin),
) -> list[ItemOut]:
"""Ein oder mehrere (count) Einzelstücke mit gemeinsamen Startwerten anlegen."""
product = _product_or_404(db, product_id)
_check_refs(db, payload.shop_id, payload.location_id)
created: list[Item] = []
for _ in range(payload.count):
item = Item(
uid=generate_uid(db),
product_id=product.id,
location_id=payload.location_id,
shop_id=payload.shop_id,
acquired_on=payload.acquired_on,
warranty_until=payload.warranty_until,
note=payload.note,
)
db.add(item)
db.flush()
created.append(item)
db.commit()
for item in created:
db.refresh(item)
return [item_to_out(i) for i in created]
@router.get("/items/by-uid/{uid}", response_model=ItemOut)
def item_by_uid(
uid: str,
db: Session = Depends(get_db),
_: User = Depends(get_current_user),
) -> ItemOut:
"""QR-Auflösung: UID → Einzelstück (mit Produktangaben)."""
item = db.query(Item).filter(Item.uid == uid.strip().upper()).first()
if item is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Kein Einzelstück mit dieser UID")
return item_to_out(item)
@router.get("/items/{item_id}", response_model=ItemOut)
def get_item(
item_id: int,
db: Session = Depends(get_db),
_: User = Depends(get_current_user),
) -> ItemOut:
return item_to_out(_item_or_404(db, item_id))
@router.patch("/items/{item_id}", response_model=ItemOut)
def update_item(
item_id: int,
payload: ItemUpdate,
db: Session = Depends(get_db),
_: User = Depends(require_admin),
) -> ItemOut:
item = _item_or_404(db, item_id)
data = payload.model_dump(exclude_unset=True)
_check_refs(db, data.get("shop_id"), data.get("location_id"))
for field, value in data.items():
setattr(item, field, value)
db.commit()
db.refresh(item)
return item_to_out(item)
@router.post("/items/{item_id}/remove", status_code=status.HTTP_204_NO_CONTENT)
def remove_item(
item_id: int,
payload: ItemRemove,
db: Session = Depends(get_db),
user: User = Depends(require_admin),
) -> None:
"""Einzelstück mit Grund entfernen als Bewegung protokolliert (Statistik je Modell)."""
item = _item_or_404(db, item_id)
db.add(
Movement(
product_id=item.product_id,
lot_id=None,
user_id=user.id,
type=MovementType.out,
quantity=1,
unit_used="Stück",
note=(payload.note or f"Einzelstück {item.uid}"),
location_id=item.location_id,
reason=payload.reason.value,
)
)
db.delete(item)
db.commit()
@router.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(
item_id: int,
db: Session = Depends(get_db),
_: User = Depends(require_admin),
) -> None:
"""Einzelstück ohne Grund löschen (Korrektur)."""
db.delete(_item_or_404(db, item_id))
db.commit()

View File

@@ -335,6 +335,7 @@ def create_product(
min_stock_in_packages=bool(payload.min_stock_in_packages),
shop_id=payload.shop_id,
product_url=payload.product_url or None,
individual=bool(payload.individual),
source="manual",
)
db.add(product)