Phase 1: Datenmodell, CRUD-API und Stammdaten

FastAPI mit SQLAlchemy und SQLite, Migrationen über Alembic. Die
Endpunkte aus projekt.md für items, categories und sizes stehen samt
Reservierung; Bilder und Anmeldung folgen in Phase 2 und 3.

Umgesetzt wie in Plan.md festgelegt:

- Feste Wertelisten als Enum UND als CHECK in der Datenbank. Die
  CHECK-Bedingung ist der eigentliche Schutz: an SQLAlchemy vorbei (Import,
  sqlite3 von Hand) käme sonst "Gril" durch, und die Filter griffen still
  nicht mehr.
- Status "draft" für den Stapel-Import. Entwürfe erscheinen weder in der
  Galerie noch in GET /items, solange nicht ausdrücklich status=draft
  angefragt wird, und lassen sich nicht reservieren.
- Titel ist freiwillig und wird sonst beim ANZEIGEN aus Kategorie und
  Grösse gebildet ("Jacken 98/104") - nicht beim Speichern, damit er einer
  späteren Korrektur der Grösse folgt.
- reservation_token: nur wer es hat, kann die eigene Reservierung aufheben.
  Verglichen mit compare_digest, gelöscht beim Freigeben und beim
  Erledigen, damit ein alter Link nicht später eine fremde Reservierung
  aufhebt.
- Eigene sizes-Tabelle mit sort_order statt SELECT DISTINCT: sonst stünde
  "104" vor "56" und jeder Tippfehler würde zur Filteroption.
- Pagination auf GET /items, in projekt.md nicht vorgesehen.

Zwei SQLite-Eigenheiten, die leicht untergehen: foreign_keys ist
standardmässig AUS (ohne PRAGMA greift ON DELETE CASCADE nicht), und
check_same_thread muss für FastAPI abgeschaltet werden. Beides in
database.py, dazu WAL fürs gleichzeitige Lesen.

24 Tests, alle grün. Zusätzlich von Hand geprüft: Migration anwenden,
Eintrag anlegen, reservieren, ohne Token freigeben (403), mit Token
freigeben (200), Swagger UI erreichbar.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GR4bNaj9GtRu57J4Niii8o
This commit is contained in:
2026-08-29 22:00:58 +02:00
co-authored by Claude Opus 5
parent 3cf6903df5
commit 3d7b7c706a
22 changed files with 1443 additions and 0 deletions
+40
View File
@@ -0,0 +1,40 @@
[alembic]
script_location = alembic
prepend_sys_path = .
# Die URL kommt aus app/config.py (env.py holt sie dort); hier absichtlich
# leer, damit der Pfad nicht an zwei Stellen gepflegt werden muss.
sqlalchemy.url =
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARNING
handlers = console
qualname =
[logger_sqlalchemy]
level = WARNING
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
+59
View File
@@ -0,0 +1,59 @@
"""Alembic-Umgebung.
Die Datenbank-URL kommt aus app/config.py statt aus alembic.ini, damit sie
nur an einer Stelle gepflegt wird und im Container automatisch stimmt.
"""
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
from app.config import einstellungen
from app.database import Basis
# Modelle importieren, damit Alembic sie in Basis.metadata sieht.
from app import models # noqa: F401
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
config.set_main_option("sqlalchemy.url", einstellungen().datenbank_url)
target_metadata = Basis.metadata
def run_migrations_offline() -> None:
context.configure(
url=config.get_main_option("sqlalchemy.url"),
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
# SQLite kann Spalten nicht ohne Weiteres ändern; Alembic baut die
# Tabelle dann im Hintergrund neu.
render_as_batch=True,
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=True,
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
+24
View File
@@ -0,0 +1,24 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}
@@ -0,0 +1,92 @@
"""Erstes Schema: Kategorien, Groessen, Items, Bilder
Revision ID: bc7d400995d3
Revises:
Create Date: 2026-08-29 19:59:57.751289
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = 'bc7d400995d3'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('categories',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=50), nullable=False),
sa.Column('slug', sa.String(length=50), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('name'),
sa.UniqueConstraint('slug')
)
op.create_table('sizes',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('label', sa.String(length=20), nullable=False),
sa.Column('sort_order', sa.Integer(), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('label')
)
op.create_table('items',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('title', sa.String(length=100), nullable=True),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('size', sa.String(length=20), nullable=True),
sa.Column('gender', sa.String(length=10), nullable=False),
sa.Column('season', sa.String(length=15), nullable=False),
sa.Column('condition', sa.String(length=20), nullable=False),
sa.Column('status', sa.String(length=20), nullable=False),
sa.Column('category_id', sa.Integer(), nullable=True),
sa.Column('reserved_by', sa.String(length=100), nullable=True),
sa.Column('reservation_token', sa.String(length=64), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
sa.CheckConstraint("condition IN ('new', 'very_good', 'good', 'worn')", name='ck_items_condition'),
sa.CheckConstraint("gender IN ('boy', 'girl', 'unisex')", name='ck_items_gender'),
sa.CheckConstraint("season IN ('spring_summer', 'autumn_winter', 'all_year')", name='ck_items_season'),
sa.CheckConstraint("status = 'draft' OR (size IS NOT NULL AND category_id IS NOT NULL)", name='ck_items_vollstaendig_wenn_veroeffentlicht'),
sa.CheckConstraint("status = 'reserved' OR (reserved_by IS NULL AND reservation_token IS NULL)", name='ck_items_reservierung_nur_wenn_reserviert'),
sa.CheckConstraint("status IN ('draft', 'available', 'reserved', 'given_away')", name='ck_items_status'),
sa.ForeignKeyConstraint(['category_id'], ['categories.id'], ondelete='RESTRICT'),
sa.PrimaryKeyConstraint('id')
)
with op.batch_alter_table('items', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_items_category_id'), ['category_id'], unique=False)
batch_op.create_index(batch_op.f('ix_items_status'), ['status'], unique=False)
batch_op.create_index('ix_items_status_size', ['status', 'size'], unique=False)
op.create_table('item_images',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('item_id', sa.Integer(), nullable=False),
sa.Column('image_url', sa.String(length=255), nullable=False),
sa.Column('is_primary', sa.Boolean(), nullable=False),
sa.ForeignKeyConstraint(['item_id'], ['items.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id')
)
with op.batch_alter_table('item_images', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_item_images_item_id'), ['item_id'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table('item_images', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_item_images_item_id'))
op.drop_table('item_images')
with op.batch_alter_table('items', schema=None) as batch_op:
batch_op.drop_index('ix_items_status_size')
batch_op.drop_index(batch_op.f('ix_items_status'))
batch_op.drop_index(batch_op.f('ix_items_category_id'))
op.drop_table('items')
op.drop_table('sizes')
op.drop_table('categories')
# ### end Alembic commands ###
View File
+39
View File
@@ -0,0 +1,39 @@
"""Einstellungen, ausschliesslich aus Umgebungsvariablen.
Nichts davon steht im Code oder im Image: Passwort und Pfade kommen beim
Start von aussen. Ohne gesetztes Passwort bleibt der Erfassungsbereich
gesperrt (siehe Phase 3) - ein mitgeliefertes Standardpasswort wäre
öffentlich bekannt und damit keines.
"""
from functools import lru_cache
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
class Einstellungen(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
# Ablageorte. Im Container zeigen beide auf eingehängte Volumes, damit
# Datenbank und Bilder einen Neustart überleben.
datenbank_pfad: Path = Path("data/kleiderboerse.sqlite")
upload_pfad: Path = Path("uploads")
# Passwort für den Erfassungsbereich. Leer = gesperrt.
admin_passwort: str = ""
admin_passwort_hash: str = ""
# Grösse eines einzelnen Bild-Uploads. Greift zusätzlich zur Prüfung in
# images.py; ein Handyfoto liegt heute bei 3-12 MB.
max_upload_bytes: int = 10 * 1024 * 1024
@property
def datenbank_url(self) -> str:
return f"sqlite:///{self.datenbank_pfad}"
@lru_cache
def einstellungen() -> Einstellungen:
"""Einmal lesen und merken - die Werte ändern sich zur Laufzeit nicht."""
return Einstellungen()
+194
View File
@@ -0,0 +1,194 @@
"""Datenbankzugriffe.
Bewusst getrennt von den Routen: so lässt sich die Logik ohne HTTP testen,
und die Routen bleiben auf Statuscodes und Berechtigungen beschränkt.
"""
from __future__ import annotations
import secrets
from sqlalchemy import func, select
from sqlalchemy.orm import Session, selectinload
from app.models import Groesse, Item, ItemBild, Kategorie, Status, jetzt
from app.schemas import ItemAendern, ItemAnlegen
# ------------------------------------------------------------ Stammdaten ---
def kategorien(db: Session) -> list[Kategorie]:
return list(db.scalars(select(Kategorie).order_by(Kategorie.name)))
def kategorie_holen(db: Session, kategorie_id: int) -> Kategorie | None:
return db.get(Kategorie, kategorie_id)
def groessen(db: Session) -> list[Groesse]:
return list(db.scalars(select(Groesse).order_by(Groesse.sort_order, Groesse.label)))
# ----------------------------------------------------------------- Items ---
def _mit_beziehungen(stmt):
"""Kategorie und Bilder gleich mitladen.
Ohne das feuert die Übersicht pro Eintrag zwei zusätzliche Abfragen
(N+1) - bei 200 Kleidungsstücken also über 400 statt drei.
"""
return stmt.options(selectinload(Item.kategorie), selectinload(Item.bilder))
def item_holen(db: Session, item_id: int) -> Item | None:
return db.scalars(
_mit_beziehungen(select(Item).where(Item.id == item_id))
).one_or_none()
def items_suchen(
db: Session,
*,
size: str | None = None,
category_id: int | None = None,
status: str | None = None,
gender: str | None = None,
season: str | None = None,
limit: int = 50,
offset: int = 0,
) -> tuple[list[Item], int]:
bedingungen = []
if size:
bedingungen.append(Item.size == size)
if category_id is not None:
bedingungen.append(Item.category_id == category_id)
if gender:
bedingungen.append(Item.gender == gender)
if season:
bedingungen.append(Item.season == season)
if status:
bedingungen.append(Item.status == status)
else:
# Entwürfe stammen aus dem Stapel-Import und haben noch keine
# Angaben. Sie gehören nicht in die Galerie - wer sie sehen will,
# muss ausdrücklich status=draft anfragen.
bedingungen.append(Item.status != Status.draft.value)
grund = select(Item)
for b in bedingungen:
grund = grund.where(b)
anzahl = db.scalar(
select(func.count()).select_from(grund.subquery())
) or 0
stmt = _mit_beziehungen(
grund.order_by(Item.created_at.desc(), Item.id.desc()).limit(limit).offset(offset)
)
return list(db.scalars(stmt)), anzahl
def item_anlegen(db: Session, daten: ItemAnlegen) -> Item:
item = Item(**daten.model_dump())
db.add(item)
db.commit()
db.refresh(item)
return item
def entwuerfe_anlegen(db: Session, anzahl: int) -> list[Item]:
"""Legt leere Entwürfe an - einen pro hochgeladenem Foto.
Wird vom Stapel-Import benutzt (Phase 2 hängt die Bilder daran). Erst
beim Nachtragen von Grösse und Kategorie wird daraus ein sichtbarer
Eintrag.
"""
neue = [Item(status=Status.draft.value) for _ in range(anzahl)]
db.add_all(neue)
db.commit()
for item in neue:
db.refresh(item)
return neue
def item_aendern(db: Session, item: Item, daten: ItemAendern) -> Item:
for feld, wert in daten.model_dump(exclude_unset=True).items():
setattr(item, feld, wert)
db.commit()
db.refresh(item)
return item
def item_loeschen(db: Session, item: Item) -> list[str]:
"""Löscht den Eintrag und meldet die Dateinamen seiner Bilder zurück.
Die Dateien selbst räumt der Aufrufer weg: ON DELETE CASCADE entfernt
nur die Zeilen in item_images, nicht die Bilder auf der Platte. Ohne
das würde das Upload-Verzeichnis mit Waisen volllaufen.
"""
dateien = [bild.image_url for bild in item.bilder]
db.delete(item)
db.commit()
return dateien
# --------------------------------------------------------- Reservierung ---
def reservieren(db: Session, item: Item, name: str) -> str:
"""Reserviert und gibt das Token zurück, mit dem sich das rückgängig
machen lässt. Das Token verlässt die Anwendung nur dieses eine Mal."""
token = secrets.token_urlsafe(32)
item.status = Status.reserved.value
item.reserved_by = name
item.reservation_token = token
item.updated_at = jetzt()
db.commit()
db.refresh(item)
return token
def freigeben(db: Session, item: Item) -> Item:
item.status = Status.available.value
item.reserved_by = None
# Token entfernen, sonst könnte ein alter Link später die Reservierung
# einer anderen Person aufheben.
item.reservation_token = None
item.updated_at = jetzt()
db.commit()
db.refresh(item)
return item
def als_abgegeben_markieren(db: Session, item: Item) -> Item:
item.status = Status.given_away.value
# Name und Token löschen: datensparsam, und die Galerie ist öffentlich.
item.reserved_by = None
item.reservation_token = None
item.updated_at = jetzt()
db.commit()
db.refresh(item)
return item
def token_gueltig(item: Item, token: str | None) -> bool:
"""Vergleich in konstanter Zeit - ein einfaches == verrät über die
Laufzeit, wie viele Zeichen am Anfang gestimmt haben."""
if not token or not item.reservation_token:
return False
return secrets.compare_digest(item.reservation_token, token)
# ---------------------------------------------------------------- Bilder ---
def bild_anhaengen(db: Session, item: Item, dateiname: str) -> ItemBild:
bild = ItemBild(
item_id=item.id,
image_url=dateiname,
# Das erste Bild eines Eintrags wird automatisch zum Vorschaubild.
is_primary=not item.bilder,
)
db.add(bild)
db.commit()
db.refresh(bild)
return bild
+61
View File
@@ -0,0 +1,61 @@
"""Datenbank-Anbindung.
SQLite braucht zwei Einstellungen, die man leicht übersieht:
* ``check_same_thread=False`` - FastAPI bearbeitet Anfragen in mehreren
Threads, SQLite verbietet das sonst für eine Verbindung.
* ``PRAGMA foreign_keys=ON`` - Fremdschlüssel sind in SQLite standardmässig
**aus**. Ohne diese Zeile greift ``ON DELETE CASCADE`` nicht, und beim
Löschen eines Eintrags blieben dessen Bilder als Waisen zurück.
"""
from collections.abc import Iterator
from sqlalchemy import create_engine, event
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from app.config import einstellungen
class Basis(DeclarativeBase):
pass
def engine_erzeugen(url: str | None = None):
cfg = einstellungen()
if url is None:
cfg.datenbank_pfad.parent.mkdir(parents=True, exist_ok=True)
url = cfg.datenbank_url
engine = create_engine(
url,
connect_args={"check_same_thread": False},
# Wartet, statt sofort abzubrechen, wenn parallel geschrieben wird.
pool_pre_ping=True,
)
@event.listens_for(engine, "connect")
def _sqlite_pragmas(dbapi_connection, _connection_record):
cur = dbapi_connection.cursor()
cur.execute("PRAGMA foreign_keys=ON")
# WAL: Lesen wird nicht mehr durch einen laufenden Schreibvorgang
# blockiert - bei einer Galerie, die viele gleichzeitig ansehen,
# der spürbarere Unterschied.
cur.execute("PRAGMA journal_mode=WAL")
cur.execute("PRAGMA busy_timeout=5000")
cur.close()
return engine
engine = engine_erzeugen()
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
def get_db() -> Iterator[Session]:
"""FastAPI-Abhängigkeit: eine Sitzung pro Anfrage, danach geschlossen."""
db = SessionLocal()
try:
yield db
finally:
db.close()
+45
View File
@@ -0,0 +1,45 @@
"""Einstiegspunkt der Anwendung.
FastAPI liefert später sowohl die REST-API als auch die gerenderten Seiten
aus - ein Dienst, eine Datenbank, kein getrenntes Frontend (Variante A aus
Plan.md).
"""
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.config import einstellungen
from app.database import SessionLocal, engine
from app.routers import categories, items
from app.seed import stammdaten_anlegen
@asynccontextmanager
async def lebenszyklus(_app: FastAPI):
# Die Tabellen legt Alembic an (alembic upgrade head). Hier werden nur
# die Stammdaten ergänzt, damit eine frische Datenbank sofort brauchbar
# ist - Kategorien und Grössen sind Voraussetzung fürs Erfassen.
cfg = einstellungen()
cfg.upload_pfad.mkdir(parents=True, exist_ok=True)
with SessionLocal() as db:
stammdaten_anlegen(db)
yield
engine.dispose()
app = FastAPI(
title="Kinderkleider-Börse",
description="Zu klein gewordene Kinderkleidung katalogisieren und weitergeben.",
version="0.1.0",
lifespan=lebenszyklus,
)
app.include_router(categories.router)
app.include_router(items.router)
@app.get("/health", tags=["Betrieb"])
def health():
"""Wird vom Docker-Healthcheck und von der CI benutzt."""
return {"status": "ok"}
+195
View File
@@ -0,0 +1,195 @@
"""Datenmodelle.
Abweichungen von projekt.md sind in Plan.md begründet; die wichtigsten:
* Feste Wertelisten statt freier VARCHAR - als Enum **und** als CHECK in der
Datenbank. Ohne die CHECK-Bedingung könnte an SQLAlchemy vorbei (Import,
Migration, sqlite3 von Hand) trotzdem "Gril" oder "reserviert" landen, und
die Filter griffen dann still nicht mehr.
* Status ``draft`` für den Stapel-Import: pro Foto entsteht ein Eintrag ohne
Titel, Grösse und Kategorie. Darum sind diese drei Felder NULL-fähig -
gefüllt sein müssen sie erst beim Wechsel auf ``available``.
* ``reservation_token``: nur wer es hat, kann die eigene Reservierung wieder
aufheben. Ein offener release-Endpunkt liesse jeden fremde Reservierungen
löschen.
"""
from __future__ import annotations
import enum
from datetime import datetime, timezone
from sqlalchemy import (
Boolean,
CheckConstraint,
DateTime,
ForeignKey,
Index,
Integer,
String,
Text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Basis
def jetzt() -> datetime:
"""Zeitstempel in UTC. Bewusst in der Anwendung statt per DB-Trigger,
damit es nicht von SQLite-Eigenheiten abhängt."""
return datetime.now(timezone.utc)
class Geschlecht(str, enum.Enum):
boy = "boy"
girl = "girl"
unisex = "unisex"
class Saison(str, enum.Enum):
spring_summer = "spring_summer"
autumn_winter = "autumn_winter"
all_year = "all_year"
class Zustand(str, enum.Enum):
new = "new"
very_good = "very_good"
good = "good"
worn = "worn"
class Status(str, enum.Enum):
draft = "draft"
available = "available"
reserved = "reserved"
given_away = "given_away"
def _werte(aufzaehlung: type[enum.Enum]) -> str:
"""Werteliste für eine CHECK-Bedingung, z.B. 'boy','girl','unisex'."""
return ", ".join(f"'{e.value}'" for e in aufzaehlung)
class Kategorie(Basis):
__tablename__ = "categories"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(50), nullable=False, unique=True)
slug: Mapped[str] = mapped_column(String(50), nullable=False, unique=True)
items: Mapped[list["Item"]] = relationship(back_populates="kategorie")
class Groesse(Basis):
"""Gepflegte Grössenliste.
projekt.md sieht GET /api/v1/sizes vor, aber keine Tabelle dazu. Die
Grössen aus den vorhandenen Einträgen zu ziehen (SELECT DISTINCT size)
wäre bequem, sortiert aber falsch - '104' käme vor '80', weil als Text
verglichen wird - und jeder Tippfehler würde zur Filteroption. Darum
eine eigene Liste mit ausdrücklicher Sortierung.
"""
__tablename__ = "sizes"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
label: Mapped[str] = mapped_column(String(20), nullable=False, unique=True)
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
class Item(Basis):
__tablename__ = "items"
__table_args__ = (
CheckConstraint(f"gender IN ({_werte(Geschlecht)})", name="ck_items_gender"),
CheckConstraint(f"season IN ({_werte(Saison)})", name="ck_items_season"),
CheckConstraint(f"condition IN ({_werte(Zustand)})", name="ck_items_condition"),
CheckConstraint(f"status IN ({_werte(Status)})", name="ck_items_status"),
# Ein veröffentlichter Eintrag braucht Grösse und Kategorie. Entwürfe
# dürfen beides offen lassen - genau dafür gibt es sie.
CheckConstraint(
"status = 'draft' OR (size IS NOT NULL AND category_id IS NOT NULL)",
name="ck_items_vollstaendig_wenn_veroeffentlicht",
),
# Nur ein reservierter Eintrag trägt einen Namen und ein Token.
CheckConstraint(
"status = 'reserved' OR (reserved_by IS NULL AND reservation_token IS NULL)",
name="ck_items_reservierung_nur_wenn_reserviert",
),
Index("ix_items_status_size", "status", "size"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
# Freiwillig: bleibt er leer, wird beim Anzeigen einer aus Kategorie und
# Grösse gebildet (siehe anzeige_titel). Bewusst nicht beim Speichern -
# sonst bliebe "Hose 98/104" stehen, wenn später die Grösse korrigiert
# wird.
title: Mapped[str | None] = mapped_column(String(100), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
size: Mapped[str | None] = mapped_column(String(20), nullable=True)
gender: Mapped[str] = mapped_column(
String(10), nullable=False, default=Geschlecht.unisex.value
)
season: Mapped[str] = mapped_column(
String(15), nullable=False, default=Saison.all_year.value
)
condition: Mapped[str] = mapped_column(
String(20), nullable=False, default=Zustand.good.value
)
status: Mapped[str] = mapped_column(
String(20), nullable=False, default=Status.available.value, index=True
)
category_id: Mapped[int | None] = mapped_column(
ForeignKey("categories.id", ondelete="RESTRICT"), nullable=True, index=True
)
reserved_by: Mapped[str | None] = mapped_column(String(100), nullable=True)
reservation_token: Mapped[str | None] = mapped_column(String(64), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, default=jetzt
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, default=jetzt, onupdate=jetzt
)
kategorie: Mapped[Kategorie | None] = relationship(back_populates="items")
bilder: Mapped[list["ItemBild"]] = relationship(
back_populates="item",
cascade="all, delete-orphan",
order_by="ItemBild.id",
)
@property
def anzeige_titel(self) -> str:
"""Titel für die Anzeige - eigener, sonst aus Kategorie und Grösse."""
if self.title:
return self.title
teile = [self.kategorie.name if self.kategorie else "Kleidungsstück"]
if self.size:
teile.append(self.size)
return " ".join(teile)
@property
def hauptbild(self) -> "ItemBild | None":
if not self.bilder:
return None
return next((b for b in self.bilder if b.is_primary), self.bilder[0])
class ItemBild(Basis):
__tablename__ = "item_images"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
item_id: Mapped[int] = mapped_column(
ForeignKey("items.id", ondelete="CASCADE"), nullable=False, index=True
)
# Dateiname, nicht vom Upload übernommen, sondern selbst erzeugt
# (siehe images.py in Phase 2). Damit ist Pfad-Manipulation ausgeschlossen.
image_url: Mapped[str] = mapped_column(String(255), nullable=False)
is_primary: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
item: Mapped[Item] = relationship(back_populates="bilder")
View File
+22
View File
@@ -0,0 +1,22 @@
"""Stammdaten: Kategorien und Grössen."""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app import crud
from app.database import get_db
from app.schemas import GroesseAus, KategorieAus
router = APIRouter(prefix="/api/v1", tags=["Stammdaten"])
@router.get("/categories", response_model=list[KategorieAus])
def kategorien_auflisten(db: Session = Depends(get_db)):
return crud.kategorien(db)
@router.get("/sizes", response_model=list[GroesseAus])
def groessen_auflisten(db: Session = Depends(get_db)):
"""Gepflegte Liste statt SELECT DISTINCT über die Einträge - so stimmt
die Reihenfolge (86 vor 104) und Tippfehler werden nicht zu Filtern."""
return crud.groessen(db)
+164
View File
@@ -0,0 +1,164 @@
"""Kleidungsstücke: durchsuchen, verwalten, reservieren.
Der Schreibschutz (Anmeldung, Rate-Limit) kommt in Phase 3 dazu; die
Berechtigungslogik der Reservierung steckt aber schon hier, weil sie den
Datenfluss bestimmt.
"""
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app import crud
from app.database import get_db
from app.models import Geschlecht, Item, Saison, Status
from app.schemas import (
ItemAendern,
ItemAnlegen,
ItemAus,
ItemListeAus,
ReservierenAus,
ReservierenRein,
)
router = APIRouter(prefix="/api/v1/items", tags=["Kleidungsstücke"])
def _item_oder_404(db: Session, item_id: int) -> Item:
item = crud.item_holen(db, item_id)
if item is None:
raise HTTPException(status_code=404, detail="Kleidungsstück nicht gefunden.")
return item
def _als_antwort(item: Item) -> ItemAus:
"""Baut die Antwort und ergänzt den abgeleiteten Titel.
display_title entsteht hier und nicht beim Speichern: sonst bliebe ein
automatisch gesetzter Titel stehen, wenn später die Grösse korrigiert
wird.
"""
return ItemAus.model_validate({**item.__dict__, "display_title": item.anzeige_titel,
"kategorie": item.kategorie, "bilder": item.bilder})
@router.get("", response_model=ItemListeAus)
def items_auflisten(
db: Session = Depends(get_db),
size: str | None = None,
category_id: int | None = None,
item_status: Status | None = Query(default=None, alias="status"),
gender: Geschlecht | None = None,
season: Saison | None = None,
limit: int = Query(default=50, ge=1, le=200),
offset: int = Query(default=0, ge=0),
):
treffer, gesamt = crud.items_suchen(
db,
size=size,
category_id=category_id,
status=item_status.value if item_status else None,
gender=gender.value if gender else None,
season=season.value if season else None,
limit=limit,
offset=offset,
)
return ItemListeAus(
items=[_als_antwort(i) for i in treffer],
total=gesamt,
limit=limit,
offset=offset,
)
@router.get("/{item_id}", response_model=ItemAus)
def item_anzeigen(item_id: int, db: Session = Depends(get_db)):
return _als_antwort(_item_oder_404(db, item_id))
@router.post("", response_model=ItemAus, status_code=status.HTTP_201_CREATED)
def item_anlegen(daten: ItemAnlegen, db: Session = Depends(get_db)):
if daten.category_id is not None and crud.kategorie_holen(db, daten.category_id) is None:
raise HTTPException(status_code=422, detail="Kategorie gibt es nicht.")
try:
item = crud.item_anlegen(db, daten)
except IntegrityError as e:
db.rollback()
# Greift z.B., wenn ein Eintrag ohne Grösse oder Kategorie
# veröffentlicht werden soll (CHECK in models.py).
raise HTTPException(
status_code=422,
detail="Ein sichtbarer Eintrag braucht Grösse und Kategorie.",
) from e
return _als_antwort(item)
@router.patch("/{item_id}", response_model=ItemAus)
def item_aendern(item_id: int, daten: ItemAendern, db: Session = Depends(get_db)):
item = _item_oder_404(db, item_id)
if daten.category_id is not None and crud.kategorie_holen(db, daten.category_id) is None:
raise HTTPException(status_code=422, detail="Kategorie gibt es nicht.")
try:
geaendert = crud.item_aendern(db, item, daten)
except IntegrityError as e:
db.rollback()
raise HTTPException(
status_code=422,
detail="Ein sichtbarer Eintrag braucht Grösse und Kategorie.",
) from e
return _als_antwort(geaendert)
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def item_loeschen(item_id: int, db: Session = Depends(get_db)):
item = _item_oder_404(db, item_id)
# Die Dateien selbst löscht Phase 2 (images.py); hier fallen nur die
# Namen an, damit die Aufräumlogik an einer Stelle liegt.
crud.item_loeschen(db, item)
return Response(status_code=status.HTTP_204_NO_CONTENT)
# --------------------------------------------------------- Reservierung ---
@router.post("/{item_id}/reserve", response_model=ReservierenAus)
def item_reservieren(item_id: int, daten: ReservierenRein, db: Session = Depends(get_db)):
item = _item_oder_404(db, item_id)
if item.status == Status.draft.value:
raise HTTPException(status_code=404, detail="Kleidungsstück nicht gefunden.")
if item.status != Status.available.value:
raise HTTPException(
status_code=409, detail="Dieses Kleidungsstück ist nicht mehr verfügbar."
)
token = crud.reservieren(db, item, daten.reserved_by)
return ReservierenAus(item=_als_antwort(item), reservation_token=token)
@router.post("/{item_id}/release", response_model=ItemAus)
def item_freigeben(
item_id: int,
token: str | None = Query(default=None),
db: Session = Depends(get_db),
ist_betreiber: bool = False, # wird in Phase 3 durch die Anmeldung ersetzt
):
"""Hebt eine Reservierung auf.
Erlaubt für den Reservierenden (mit Token) und den Betreiber. Ohne
Token wäre der Endpunkt offen - dann könnte jeder Besucher fremde
Reservierungen löschen, und weil reserved_by mitgeht, bliebe nicht
einmal nachvollziehbar, dass jemand reserviert hatte.
"""
item = _item_oder_404(db, item_id)
if item.status != Status.reserved.value:
raise HTTPException(status_code=409, detail="Dieses Kleidungsstück ist nicht reserviert.")
if not ist_betreiber and not crud.token_gueltig(item, token):
raise HTTPException(
status_code=403,
detail="Zum Aufheben braucht es den Link aus der Reservierung.",
)
return _als_antwort(crud.freigeben(db, item))
@router.post("/{item_id}/mark-given", response_model=ItemAus)
def item_abgegeben(item_id: int, db: Session = Depends(get_db)):
item = _item_oder_404(db, item_id)
return _als_antwort(crud.als_abgegeben_markieren(db, item))
+145
View File
@@ -0,0 +1,145 @@
"""Pydantic-Schemas für Anfragen und Antworten.
Die Aufteilung in Anlegen / Ändern / Antwort ist Absicht: beim Ändern sind
alle Felder freiwillig (PATCH), beim Anlegen gelten die Pflichtfelder, und
in der Antwort dürfen Felder stehen, die niemand setzen darf - etwa der
abgeleitete Titel oder das Reservierungs-Token.
"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.models import Geschlecht, Saison, Status, Zustand
class KategorieAus(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
slug: str
class GroesseAus(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
label: str
sort_order: int
class BildAus(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
image_url: str
is_primary: bool
def _leer_zu_none(wert: str | None) -> str | None:
"""Leere Eingaben aus einem Formular als 'nicht gesetzt' behandeln.
Ein leeres Titelfeld soll NULL ergeben, damit der abgeleitete Titel
greift - nicht den leeren String, der später als "gesetzt" gälte.
"""
if wert is None:
return None
wert = wert.strip()
return wert or None
class ItemAnlegen(BaseModel):
title: str | None = Field(default=None, max_length=100)
description: str | None = None
size: str | None = Field(default=None, max_length=20)
gender: Geschlecht = Geschlecht.unisex
season: Saison = Saison.all_year
condition: Zustand = Zustand.good
category_id: int | None = None
status: Status = Status.available
_titel_putzen = field_validator("title", "size", "description", mode="before")(
lambda v: _leer_zu_none(v) if isinstance(v, str) or v is None else v
)
class ItemAendern(BaseModel):
"""Alle Felder freiwillig - nicht gesetzte bleiben unverändert.
Bewusst ohne reserved_by/reservation_token: eine Reservierung läuft
ausschliesslich über die eigenen Endpunkte, sonst könnte man sie per
PATCH an Rate-Limit und Token-Prüfung vorbei setzen.
"""
title: str | None = Field(default=None, max_length=100)
description: str | None = None
size: str | None = Field(default=None, max_length=20)
gender: Geschlecht | None = None
season: Saison | None = None
condition: Zustand | None = None
category_id: int | None = None
status: Status | None = None
_putzen = field_validator("title", "size", "description", mode="before")(
lambda v: _leer_zu_none(v) if isinstance(v, str) or v is None else v
)
class ItemAus(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
title: str | None
display_title: str
description: str | None
size: str | None
gender: Geschlecht
season: Saison
condition: Zustand
status: Status
category_id: int | None
kategorie: KategorieAus | None = None
reserved_by: str | None
created_at: datetime
updated_at: datetime
bilder: list[BildAus] = []
class ItemListeAus(BaseModel):
"""Antwort der Übersicht - mit Seitenangaben.
projekt.md sieht keine Pagination vor; ohne sie liefert der Endpunkt
irgendwann den gesamten Bestand samt Bildpfaden in einer Antwort.
"""
items: list[ItemAus]
total: int
limit: int
offset: int
class ReservierenRein(BaseModel):
reserved_by: str = Field(min_length=1, max_length=100)
@field_validator("reserved_by")
@classmethod
def keine_links(cls, wert: str) -> str:
wert = wert.strip()
if not wert:
raise ValueError("Bitte einen Namen angeben.")
# Die Galerie ist frei zugänglich; ohne diese Bremse wird das Feld
# früher oder später als Werbefläche benutzt.
verdaechtig = ("http://", "https://", "www.", "<")
if any(teil in wert.lower() for teil in verdaechtig):
raise ValueError("Bitte nur einen Namen angeben, keine Links.")
return wert
class ReservierenAus(BaseModel):
item: ItemAus
# Nur unmittelbar nach dem Reservieren sichtbar: daraus wird der Link
# zum Aufheben gebildet. Taucht in keiner anderen Antwort auf.
reservation_token: str
+50
View File
@@ -0,0 +1,50 @@
"""Stammdaten vorbefüllen: Kategorien und Grössen.
Läuft beim Start und legt nur an, was fehlt - so gehen von Hand ergänzte
Einträge bei einem Neustart nicht verloren.
"""
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models import Groesse, Kategorie
KATEGORIEN = [
("Hosen", "hosen"),
("Oberteile", "oberteile"),
("Pullover", "pullover"),
("Jacken", "jacken"),
("Kleider & Röcke", "kleider-roecke"),
("Schuhe", "schuhe"),
("Unterwäsche & Socken", "unterwaesche-socken"),
("Schlafanzüge", "schlafanzuege"),
("Mützen & Handschuhe", "muetzen-handschuhe"),
("Sonstiges", "sonstiges"),
]
# Reihenfolge ausdrücklich vergeben. Als Text sortiert stünde "104" vor
# "56" - genau der Grund für die eigene Tabelle.
GROESSEN = [
# Babys und Kleinkinder (Körpergrösse in cm)
*[(str(cm), i) for i, cm in enumerate(range(50, 105, 6), start=10)],
# Kinder
*[(f"{cm}/{cm + 6}", 100 + i) for i, cm in enumerate(range(98, 170, 12))],
*[(str(cm), 200 + i) for i, cm in enumerate(range(104, 182, 6))],
# Schuhe
*[(f"Schuh {nr}", 300 + i) for i, nr in enumerate(range(16, 41))],
]
def stammdaten_anlegen(db: Session) -> None:
vorhanden = set(db.scalars(select(Kategorie.slug)))
for name, slug in KATEGORIEN:
if slug not in vorhanden:
db.add(Kategorie(name=name, slug=slug))
bekannt = set(db.scalars(select(Groesse.label)))
for label, sort_order in GROESSEN:
if label not in bekannt:
db.add(Groesse(label=label, sort_order=sort_order))
bekannt.add(label)
db.commit()
+3
View File
@@ -0,0 +1,3 @@
-r requirements.txt
pytest>=8.3,<9
httpx>=0.27,<0.29
+8
View File
@@ -0,0 +1,8 @@
fastapi>=0.115,<0.116
uvicorn[standard]>=0.32,<0.35
sqlalchemy>=2.0,<2.1
alembic>=1.14,<1.15
pydantic>=2.9,<3
pydantic-settings>=2.6,<3
python-multipart>=0.0.17
jinja2>=3.1,<4
View File
+45
View File
@@ -0,0 +1,45 @@
"""Gemeinsame Test-Vorbereitung.
Jeder Test bekommt eine eigene, frische SQLite-Datei in einem temporären
Verzeichnis - keine geteilten Zustände zwischen Tests, und die echte
Datenbank unter data/ wird nie angefasst.
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import sessionmaker
from app.database import Basis, engine_erzeugen, get_db
from app.main import app
from app.seed import stammdaten_anlegen
@pytest.fixture
def db_sitzung(tmp_path):
engine = engine_erzeugen(f"sqlite:///{tmp_path/'test.sqlite'}")
Basis.metadata.create_all(engine)
Sitzung = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
with Sitzung() as db:
stammdaten_anlegen(db)
yield db
engine.dispose()
@pytest.fixture
def client(db_sitzung):
"""TestClient, der auf dieselbe Testdatenbank zeigt wie db_sitzung."""
def _db():
yield db_sitzung
app.dependency_overrides[get_db] = _db
# Bewusst ohne "with": nur dann startet Starlette den Lifespan, und der
# würde die echte Datenbank unter data/ anlegen und befüllen. Die
# Stammdaten stehen in der Testdatenbank ohnehin schon (db_sitzung).
yield TestClient(app)
app.dependency_overrides.clear()
@pytest.fixture
def kategorie_id(client):
return client.get("/api/v1/categories").json()[0]["id"]
+53
View File
@@ -0,0 +1,53 @@
"""Tests zum Stapel-Import.
Entwürfe entstehen beim Hochladen vieler Fotos auf einmal und haben noch
keine Angaben. Entscheidend ist, dass sie nicht in der Galerie auftauchen.
"""
from app import crud
def test_entwuerfe_erscheinen_nicht_in_der_galerie(client, db_sitzung, kategorie_id):
crud.entwuerfe_anlegen(db_sitzung, 3)
client.post("/api/v1/items", json={"size": "80", "category_id": kategorie_id})
galerie = client.get("/api/v1/items").json()
assert galerie["total"] == 1
entwuerfe = client.get("/api/v1/items?status=draft").json()
assert entwuerfe["total"] == 3
def test_entwurf_darf_unvollstaendig_sein(client, db_sitzung):
entwurf = crud.entwuerfe_anlegen(db_sitzung, 1)[0]
assert entwurf.size is None
assert entwurf.category_id is None
assert entwurf.anzeige_titel == "Kleidungsstück"
def test_entwurf_wird_durch_nachtragen_sichtbar(client, db_sitzung, kategorie_id):
entwurf = crud.entwuerfe_anlegen(db_sitzung, 1)[0]
antwort = client.patch(
f"/api/v1/items/{entwurf.id}",
json={"size": "92", "category_id": kategorie_id, "status": "available"},
)
assert antwort.status_code == 200
assert client.get("/api/v1/items").json()["total"] == 1
def test_unvollstaendiger_eintrag_kann_nicht_veroeffentlicht_werden(
client, db_sitzung
):
"""Die CHECK-Bedingung greift auch dann, wenn die Route es zuliesse."""
entwurf = crud.entwuerfe_anlegen(db_sitzung, 1)[0]
antwort = client.patch(f"/api/v1/items/{entwurf.id}", json={"status": "available"})
assert antwort.status_code == 422
def test_entwurf_kann_nicht_reserviert_werden(client, db_sitzung):
entwurf = crud.entwuerfe_anlegen(db_sitzung, 1)[0]
antwort = client.post(
f"/api/v1/items/{entwurf.id}/reserve", json={"reserved_by": "Wer auch immer"}
)
assert antwort.status_code == 404
+119
View File
@@ -0,0 +1,119 @@
"""Tests für Anlegen, Filtern, Ändern und Löschen."""
def test_stammdaten_sind_da(client):
kategorien = client.get("/api/v1/categories").json()
groessen = client.get("/api/v1/sizes").json()
assert len(kategorien) >= 10
assert len(groessen) > 20
def test_groessen_sind_sinnvoll_sortiert(client):
"""Der eigentliche Grund für die eigene Tabelle: als Text sortiert
stünde '104' vor '56'."""
labels = [g["label"] for g in client.get("/api/v1/sizes").json()]
assert labels.index("56") < labels.index("104")
def test_anlegen_und_abrufen(client, kategorie_id):
antwort = client.post(
"/api/v1/items",
json={"title": "Warme Winterjacke rot", "size": "98/104",
"category_id": kategorie_id, "gender": "girl"},
)
assert antwort.status_code == 201
item = antwort.json()
assert item["title"] == "Warme Winterjacke rot"
assert item["display_title"] == "Warme Winterjacke rot"
assert item["status"] == "available"
einzeln = client.get(f"/api/v1/items/{item['id']}").json()
assert einzeln["kategorie"]["id"] == kategorie_id
def test_titel_wird_aus_kategorie_und_groesse_gebildet(client, kategorie_id):
"""Titel ist freiwillig - ohne ihn entsteht 'Hosen 98/104'."""
item = client.post(
"/api/v1/items",
json={"size": "98/104", "category_id": kategorie_id},
).json()
assert item["title"] is None
kategorie_name = client.get("/api/v1/categories").json()[0]["name"]
assert item["display_title"] == f"{kategorie_name} 98/104"
def test_abgeleiteter_titel_folgt_einer_korrektur(client, kategorie_id):
"""Deshalb wird er beim Anzeigen gebildet und nicht beim Speichern."""
item = client.post(
"/api/v1/items", json={"size": "98/104", "category_id": kategorie_id}
).json()
geaendert = client.patch(
f"/api/v1/items/{item['id']}", json={"size": "110/116"}
).json()
assert geaendert["display_title"].endswith("110/116")
def test_leerer_titel_zaehlt_als_nicht_gesetzt(client, kategorie_id):
item = client.post(
"/api/v1/items",
json={"title": " ", "size": "80", "category_id": kategorie_id},
).json()
assert item["title"] is None
assert item["display_title"].endswith("80")
def test_filtern(client, kategorie_id):
for groesse, geschlecht in [("80", "boy"), ("86", "girl"), ("80", "girl")]:
client.post(
"/api/v1/items",
json={"size": groesse, "category_id": kategorie_id, "gender": geschlecht},
)
assert client.get("/api/v1/items?size=80").json()["total"] == 2
assert client.get("/api/v1/items?gender=girl").json()["total"] == 2
assert client.get("/api/v1/items?size=80&gender=girl").json()["total"] == 1
def test_pagination(client, kategorie_id):
for i in range(5):
client.post(
"/api/v1/items",
json={"size": "80", "category_id": kategorie_id, "title": f"Teil {i}"},
)
seite = client.get("/api/v1/items?limit=2&offset=0").json()
assert seite["total"] == 5
assert len(seite["items"]) == 2
def test_aendern(client, kategorie_id):
item = client.post(
"/api/v1/items", json={"size": "80", "category_id": kategorie_id}
).json()
geaendert = client.patch(
f"/api/v1/items/{item['id']}", json={"condition": "worn"}
).json()
assert geaendert["condition"] == "worn"
# Nicht gesetzte Felder bleiben unangetastet
assert geaendert["size"] == "80"
def test_loeschen(client, kategorie_id):
item = client.post(
"/api/v1/items", json={"size": "80", "category_id": kategorie_id}
).json()
assert client.delete(f"/api/v1/items/{item['id']}").status_code == 204
assert client.get(f"/api/v1/items/{item['id']}").status_code == 404
def test_unbekannte_kategorie_wird_abgewiesen(client):
antwort = client.post("/api/v1/items", json={"size": "80", "category_id": 9999})
assert antwort.status_code == 422
def test_ungueltiger_wert_wird_abgewiesen(client, kategorie_id):
"""Feste Wertelisten statt freier VARCHAR - 'Gril' darf nicht durch."""
antwort = client.post(
"/api/v1/items",
json={"size": "80", "category_id": kategorie_id, "gender": "Gril"},
)
assert antwort.status_code == 422
+85
View File
@@ -0,0 +1,85 @@
"""Tests rund ums Reservieren.
Schwerpunkt ist die Berechtigung: die Galerie ist frei zugänglich, also
darf ein offener release-Endpunkt nicht fremde Reservierungen löschen
können.
"""
import pytest
@pytest.fixture
def item_id(client, kategorie_id):
return client.post(
"/api/v1/items", json={"size": "98/104", "category_id": kategorie_id}
).json()["id"]
def test_reservieren(client, item_id):
antwort = client.post(
f"/api/v1/items/{item_id}/reserve", json={"reserved_by": "Familie Meier"}
)
assert antwort.status_code == 200
daten = antwort.json()
assert daten["item"]["status"] == "reserved"
assert daten["item"]["reserved_by"] == "Familie Meier"
assert len(daten["reservation_token"]) > 20
def test_zweimal_reservieren_geht_nicht(client, item_id):
client.post(f"/api/v1/items/{item_id}/reserve", json={"reserved_by": "A"})
zweiter = client.post(f"/api/v1/items/{item_id}/reserve", json={"reserved_by": "B"})
assert zweiter.status_code == 409
def test_freigeben_mit_token(client, item_id):
token = client.post(
f"/api/v1/items/{item_id}/reserve", json={"reserved_by": "Familie Meier"}
).json()["reservation_token"]
antwort = client.post(f"/api/v1/items/{item_id}/release?token={token}")
assert antwort.status_code == 200
assert antwort.json()["status"] == "available"
assert antwort.json()["reserved_by"] is None
def test_freigeben_ohne_token_wird_abgewiesen(client, item_id):
"""Der Kern der Entscheidung: sonst löscht jeder Besucher fremde
Reservierungen."""
client.post(f"/api/v1/items/{item_id}/reserve", json={"reserved_by": "Familie Meier"})
assert client.post(f"/api/v1/items/{item_id}/release").status_code == 403
assert client.post(f"/api/v1/items/{item_id}/release?token=falsch").status_code == 403
# Reservierung steht unverändert
item = client.get(f"/api/v1/items/{item_id}").json()
assert item["status"] == "reserved"
assert item["reserved_by"] == "Familie Meier"
def test_token_gilt_nach_dem_freigeben_nicht_mehr(client, item_id):
"""Sonst könnte ein alter Link später die Reservierung einer anderen
Person aufheben."""
token = client.post(
f"/api/v1/items/{item_id}/reserve", json={"reserved_by": "Erste"}
).json()["reservation_token"]
client.post(f"/api/v1/items/{item_id}/release?token={token}")
client.post(f"/api/v1/items/{item_id}/reserve", json={"reserved_by": "Zweite"})
assert client.post(f"/api/v1/items/{item_id}/release?token={token}").status_code == 403
def test_abgegeben_loescht_den_namen(client, item_id):
"""Datensparsam - die Galerie ist öffentlich."""
client.post(f"/api/v1/items/{item_id}/reserve", json={"reserved_by": "Familie Meier"})
antwort = client.post(f"/api/v1/items/{item_id}/mark-given").json()
assert antwort["status"] == "given_away"
assert antwort["reserved_by"] is None
def test_name_ohne_werbung(client, item_id):
antwort = client.post(
f"/api/v1/items/{item_id}/reserve",
json={"reserved_by": "Jetzt billig kaufen http://spam.example"},
)
assert antwort.status_code == 422