Backend: globale Einzelstueck-Liste (GET /items) + Bild-Version an Produkten
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>
This commit is contained in:
@@ -7,7 +7,7 @@ from datetime import date
|
|||||||
from fastapi import HTTPException, status
|
from fastapi import HTTPException, status
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
from .models import Barcode, Category, CategoryTracking, Lot, Product
|
from .models import Barcode, Category, CategoryTracking, Lot, Product, ProductImage
|
||||||
from .schemas import BarcodeOut, LocationMinStockOut, ProductOut
|
from .schemas import BarcodeOut, LocationMinStockOut, ProductOut
|
||||||
from .services.conversion import KIND_OF_BASE, display_unit_info
|
from .services.conversion import KIND_OF_BASE, display_unit_info
|
||||||
from .services.stock import current_stock
|
from .services.stock import current_stock
|
||||||
@@ -44,6 +44,14 @@ def product_to_out(db: Session, product: Product) -> ProductOut:
|
|||||||
for b in db.query(Barcode).filter(Barcode.product_id == product.id).order_by(Barcode.id).all()
|
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
|
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.tracking = CategoryTracking(product_tracking(db, product))
|
||||||
out.shop_name = product.shop.name if product.shop else None
|
out.shop_name = product.shop.name if product.shop else None
|
||||||
out.kind = KIND_OF_BASE[product.base_unit].value
|
out.kind = KIND_OF_BASE[product.base_unit].value
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ import re
|
|||||||
|
|
||||||
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
|
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
|
||||||
from fastapi.responses import Response
|
from fastapi.responses import Response
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session, joinedload
|
||||||
|
|
||||||
from ..database import get_db
|
from ..database import get_db
|
||||||
from ..deps import get_current_user, require_admin
|
from ..deps import get_current_user, require_admin
|
||||||
@@ -115,6 +115,30 @@ def create_items(
|
|||||||
return [item_to_out(i) for i in created]
|
return [item_to_out(i) for i in created]
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/items", response_model=list[ItemOut])
|
||||||
|
def list_all_items(
|
||||||
|
db: Session = Depends(get_db),
|
||||||
|
_: User = Depends(get_current_user),
|
||||||
|
) -> list[ItemOut]:
|
||||||
|
"""Alle Einzelstücke über alle Produkte – Grundlage der Einzelstück-Liste.
|
||||||
|
|
||||||
|
Angereichert mit Produkt-, Kategorie-, Lagerort- und Shop-Angaben; die
|
||||||
|
Beziehungen werden vorgeladen, damit die Liste nicht in N+1-Abfragen zerfällt.
|
||||||
|
"""
|
||||||
|
rows = (
|
||||||
|
db.query(Item)
|
||||||
|
.options(
|
||||||
|
joinedload(Item.product).joinedload(Product.category),
|
||||||
|
joinedload(Item.location),
|
||||||
|
joinedload(Item.shop),
|
||||||
|
joinedload(Item.documents),
|
||||||
|
)
|
||||||
|
.order_by(Item.id)
|
||||||
|
.all()
|
||||||
|
)
|
||||||
|
return [item_to_out(i) for i in rows]
|
||||||
|
|
||||||
|
|
||||||
@router.get("/items/by-uid/{uid}", response_model=ItemOut)
|
@router.get("/items/by-uid/{uid}", response_model=ItemOut)
|
||||||
def item_by_uid(
|
def item_by_uid(
|
||||||
uid: str,
|
uid: str,
|
||||||
|
|||||||
@@ -354,6 +354,9 @@ class ProductOut(BaseModel):
|
|||||||
# angereichert:
|
# angereichert:
|
||||||
stock: float = 0.0
|
stock: float = 0.0
|
||||||
expired_count: int = 0
|
expired_count: int = 0
|
||||||
|
# Epoch (Sekunden) der letzten Bildänderung, sonst None (= kein Bild). Der
|
||||||
|
# Client cacht Bilder danach und lädt nur bei geänderter Version neu.
|
||||||
|
image_version: int | None = None
|
||||||
kind: str = ""
|
kind: str = ""
|
||||||
unit_name: str = ""
|
unit_name: str = ""
|
||||||
unit_factor: float = 1.0
|
unit_factor: float = 1.0
|
||||||
@@ -591,6 +594,8 @@ class ItemOut(BaseModel):
|
|||||||
# Für QR-Auflösung/Anzeige mitgeliefert:
|
# Für QR-Auflösung/Anzeige mitgeliefert:
|
||||||
product_name: str | None = None
|
product_name: str | None = None
|
||||||
product_brand: str | None = None
|
product_brand: str | None = None
|
||||||
|
category_id: int | None = None
|
||||||
|
category_name: str | None = None
|
||||||
documents: list[ItemDocumentOut] = []
|
documents: list[ItemDocumentOut] = []
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -28,4 +28,7 @@ def item_to_out(item: Item) -> ItemOut:
|
|||||||
out.shop_name = item.shop.name if item.shop else None
|
out.shop_name = item.shop.name if item.shop else None
|
||||||
out.product_name = item.product.name if item.product else None
|
out.product_name = item.product.name if item.product else None
|
||||||
out.product_brand = item.product.brand if item.product else None
|
out.product_brand = item.product.brand if item.product else None
|
||||||
|
kategorie = item.product.category if item.product else None
|
||||||
|
out.category_id = kategorie.id if kategorie else None
|
||||||
|
out.category_name = kategorie.name if kategorie else None
|
||||||
return out
|
return out
|
||||||
|
|||||||
@@ -2,8 +2,17 @@
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from app.models import Category, CategoryTracking, Item, ItemDocument, Product, Role, User
|
from app.models import (
|
||||||
from app.routers.items import _safe_filename, create_items, get_item
|
Category,
|
||||||
|
CategoryTracking,
|
||||||
|
Item,
|
||||||
|
ItemDocument,
|
||||||
|
Location,
|
||||||
|
Product,
|
||||||
|
Role,
|
||||||
|
User,
|
||||||
|
)
|
||||||
|
from app.routers.items import _safe_filename, create_items, get_item, list_all_items
|
||||||
from app.schemas import ItemCreate
|
from app.schemas import ItemCreate
|
||||||
|
|
||||||
|
|
||||||
@@ -49,6 +58,28 @@ def test_item_out_listet_belege(db, admin):
|
|||||||
assert fresh.documents[0].content_type == "application/pdf"
|
assert fresh.documents[0].content_type == "application/pdf"
|
||||||
|
|
||||||
|
|
||||||
|
def test_list_all_items_angereichert(db, admin):
|
||||||
|
cat = Category(name="Elektronik", tracking=CategoryTracking.object.value)
|
||||||
|
db.add(cat)
|
||||||
|
db.flush()
|
||||||
|
product = Product(name="Kamera", brand="Sony", category_id=cat.id, individual=True)
|
||||||
|
db.add(product)
|
||||||
|
db.flush()
|
||||||
|
ort = Location(name="Keller")
|
||||||
|
db.add(ort)
|
||||||
|
db.commit()
|
||||||
|
create_items(product.id, ItemCreate(count=2, location_id=ort.id), db=db, _=admin)
|
||||||
|
|
||||||
|
alle = list_all_items(db=db, _=admin)
|
||||||
|
assert len(alle) == 2
|
||||||
|
e = alle[0]
|
||||||
|
assert e.product_name == "Kamera"
|
||||||
|
assert e.product_brand == "Sony"
|
||||||
|
assert e.category_name == "Elektronik"
|
||||||
|
assert e.location_id == ort.id
|
||||||
|
assert e.location_name == "Keller"
|
||||||
|
|
||||||
|
|
||||||
def test_safe_filename_entfernt_header_zeichen():
|
def test_safe_filename_entfernt_header_zeichen():
|
||||||
assert _safe_filename('a"b\r\nc.pdf') == "abc.pdf"
|
assert _safe_filename('a"b\r\nc.pdf') == "abc.pdf"
|
||||||
assert _safe_filename(" ") == "beleg"
|
assert _safe_filename(" ") == "beleg"
|
||||||
|
|||||||
@@ -17,6 +17,18 @@ def _artikel(db, url=None):
|
|||||||
return p
|
return p
|
||||||
|
|
||||||
|
|
||||||
|
def test_image_version_spiegelt_bildzeitstempel(db):
|
||||||
|
from app.crud import product_to_out
|
||||||
|
|
||||||
|
p = _artikel(db)
|
||||||
|
assert product_to_out(db, p).image_version is None # kein Bild
|
||||||
|
img = ProductImage(product_id=p.id, content_type="image/png", data=EIN_PIXEL)
|
||||||
|
db.add(img)
|
||||||
|
db.commit()
|
||||||
|
db.refresh(img)
|
||||||
|
assert product_to_out(db, p).image_version == int(img.updated_at.timestamp())
|
||||||
|
|
||||||
|
|
||||||
def test_ohne_bildadresse_gibt_es_nichts_zu_holen(db, monkeypatch):
|
def test_ohne_bildadresse_gibt_es_nichts_zu_holen(db, monkeypatch):
|
||||||
monkeypatch.setattr(images, "fetch", lambda url: (_ for _ in ()).throw(AssertionError))
|
monkeypatch.setattr(images, "fetch", lambda url: (_ for _ in ()).throw(AssertionError))
|
||||||
assert images.ensure(db, _artikel(db)) is None
|
assert images.ensure(db, _artikel(db)) is None
|
||||||
|
|||||||
Reference in New Issue
Block a user