Zum Danebenlegen, wenn Sie einen FastAPI-Dienst bauen und sich fragen, welche Ladestrategie hier passt, wo eine Regel hingehört oder warum die Session nach dem Commit meckert. Das Blatt erklärt die Muster nicht — dafür ist das Seminar da. Maßgeblich sind die Originaldokumentationen von Pydantic, SQLAlchemy und FastAPI; Stand hier: Pydantic 2.13, SQLAlchemy 2.0, Python 3.14.
Pydantic v1 → v2
| Pydantic v1 | Pydantic v2 |
|---|---|
parse_obj(...) | model_validate(...) |
.dict() | .model_dump() |
.json() | .model_dump_json() |
innere class Config | model_config = ConfigDict(...) |
@validator | @field_validator |
@root_validator | @model_validator |
Die v1-Namen existieren teils noch als Alias — in neuem Code nicht verwenden. Die Validator-Signaturen haben sich geändert, v1-Wissen überträgt sich nicht eins zu eins.
Modul: Schemas mit Pydantic v2
Aliase: interner Name gegen Außenvertrag
| Config-Schlüssel | Vorgabe | Wirkung |
|---|---|---|
validate_by_alias | True | Eingang unter dem Alias erlaubt |
validate_by_name | False | Eingang unter dem Attributnamen erlaubt |
serialize_by_alias | False | Ausgabe unter dem Alias |
populate_by_name | — | seit 2.11 nicht mehr empfohlen, fällt in v3 weg |
class Position(BaseModel):
model_config = ConfigDict(validate_by_name=True)
tages_satz: float = Field(alias="tagesSatz")
Beide Validierungsschlüssel zusammen lassen den Eingang unter beiden Namen zu.
Antworten serialisiert FastAPI über response_model_by_alias von sich aus —
ein selbst aufgerufenes model_dump() braucht by_alias=True zusätzlich.
Modul: Schemas mit Pydantic v2
Regeln: welches Werkzeug wofür
| Reichweite | Werkzeug | Sieht |
|---|---|---|
| ein Feld, wiederverwendbar | Annotated[typ, Field(...)] | den Feldwert |
| ein Feld, Vorverarbeitung | BeforeValidator | den Rohwert |
| mehrere Felder | @model_validator(mode="after") | das gebaute Modell |
| abgeleiteter Wert | @computed_field + @property | das Modell, nur lesend |
@model_validator(mode="after")
def pruefe_zeitraum(self):
if self.ende < self.beginn:
raise ValueError("Ende vor Beginn")
return self # ohne return wird das Modell leer
@computed_field steht über @property, nie darunter. Berechnete Felder
sind ausgangsseitig — ein Client kann sie nicht mitschicken und damit nicht
fälschen.
Modul: Schemas mit Pydantic v2
Validieren außerhalb von HTTP
AngebotCreate.model_validate(roh) # dict, Queue, Webhook
AngebotCreate.model_validate(zeile, from_attributes=True) # ORM-Objekt
modell.model_dump(by_alias=True, mode="json") # nach außen
Ein manuelles model_validate wirft ValidationError, den FastAPI nicht
übersetzt — ohne try wird daraus 500 statt 422.
Modul: Schemas mit Pydantic v2
ORM-Modelle: wann der Zusatz nötig ist
| Fall | Schreibweise |
|---|---|
| einfache Spalte | Mapped[str] |
| Primärschlüssel | mapped_column(primary_key=True) |
| Fremdschlüssel | mapped_column(ForeignKey(...)) |
| Vorgabewert | mapped_column(default=...) |
| eindeutig | mapped_column(unique=True) |
| Beziehung | relationship(back_populates=...) |
Die Annotation trägt den Typ, mapped_column alles darüber hinaus.
DeclarativeBase ersetzt die alte Fabrikfunktion declarative_base().
Modul: Asynchrone Persistenz
Session: flush, commit, rollback
| Vorgang | Wirkung |
|---|---|
flush | SQL in die offene Transaktion; Schlüssel sind danach gesetzt |
commit | beendet die Transaktion, macht alles dauerhaft |
rollback | macht jede ausstehende Änderung gemeinsam rückgängig |
| autobegin | öffnet die Transaktion implizit bei der ersten Anweisung |
Eine Engine je Prozess, eine Session je Anfrage, eine Transaktion je
Session. expire_on_commit=False ist in Async faktisch Pflicht — sonst löst der
Zugriff nach dem Commit eine versteckte Nachladeabfrage aus und scheitert.
Modul: Asynchrone Persistenz
Ladestrategien wählen
| Strategie | Abfragen | Passt für |
|---|---|---|
| lazy (Vorgabe) | 1 + je Zugriff | in Async faktisch verboten |
selectinload | 2 | Sammlungen |
joinedload | 1 | Einzelbezüge |
subqueryload | 2 | Altcode, abgelöst |
raiseload("*") | Fehler | erzwingt die Zusage |
Die Leitfrage: Welche verbundenen Daten sind garantiert geladen, wenn diese
Abfrage zurückkommt? Ein Join auf eine Sammlung vervielfacht Zeilen und
verlangt unique() auf dem Ergebnis. Verschachtelte Pfade brauchen eine
Option je Sprung. Die Strategie gehört an die Abfrage, nicht ans Modell —
am Modell gesetzt zahlt jeder Aufrufer mit, auch wer die Sammlung nicht liest.
Modul: Ladestrategien und Produktionsreife
Vor Produktion
| Thema | Übung | Produktion |
|---|---|---|
| Schema | create_all beim Start | Alembic-Migrationen |
| Datenbank | SQLite über aiosqlite | PostgreSQL über asyncpg |
| Menge | alle Zeilen | Pagination, Aggregate |
Pool-Stellschrauben: pool_size, max_overflow, pool_timeout,
pool_pre_ping. Offset-Pagination wird auf tiefen Seiten langsam — Keyset
fragt nach Zeilen hinter der letzten ID und bleibt mit Index gleich schnell.
Modul: Ladestrategien und Produktionsreife
Dependencies: Lebensdauer und Anbringung
| Gegenstand | Lebensdauer |
|---|---|
| Engine, Session-Fabrik | Prozess (Modulebene) |
| Schema, Startbestand | Prozess (Lifespan) |
| Session, Kontext, Services | Anfrage |
| Anbringung | Wie | Wann |
|---|---|---|
| Gate | in dependencies der Route | Handler braucht den Benutzer nicht |
| Parameter | als typisiertes Argument | Handler prüft selbst weiter |
| Router | dependencies am APIRouter | Regel gilt für alle Routen darunter |
| App | globale dependencies | Mandantenprüfung und Ähnliches |
Innerhalb einer Anfrage läuft jede Dependency genau einmal; alle Consumer
bekommen dasselbe Ergebnis. Deshalb sehen zwei Services dieselbe Session und
damit dieselbe Transaktion. use_cache=False in Depends durchbricht das.
Faustregel für die Lebensdauer: Dürfen zwei gleichzeitige Anfragen sich das
gefahrlos teilen?
Modul: Abhängigkeitsgraphen komponieren · Grenzen, Autorisierung und Test
Autorisierung
| rollenbasiert | attributbasiert | |
|---|---|---|
| Frage | Wer bist du? | Was gilt für diesen Fall? |
| Vergleicht | Rolle gegen Liste | Nutzerdaten gegen Objektdaten |
| Stärke | grobe Grenzen | Einzelfallentscheidung |
401 heißt unauthentifiziert — der Server weiß nicht, wer ruft. 403 heißt
bekannt, aber nicht berechtigt. Wachsen die Rollennamen zu Gebilden wie
vertrieb-nord-nur-lesen, kodieren sie Attribute; dann eine Ebene Indirektion
einziehen und Rechte statt Rollen prüfen.
Zum Testen: app.dependency_overrides[provider] = ersatz — Schlüssel ist das
Funktionsobjekt selbst. Zwischen Tests mit .clear() aufräumen.
Modul: Grenzen, Autorisierung und Test
Nebenläufige Aufgaben starten
| Weg | Bei Fehlschlag | Eignung |
|---|---|---|
create_task allein | Fehler bleibt unbemerkt | verwaiste Aufgabe, meiden |
asyncio.gather | Geschwister laufen weiter | wenn Teilerfolg erlaubt ist |
asyncio.TaskGroup | Geschwister werden abgebrochen | wenn alles oder nichts gilt |
Rechenintensives gehört über loop.run_in_executor(None, fn, ...) in den
Threadpool — der ist gemeinsam und klein, also für kurze Spitzen, nicht als
Auftragswarteschlange. Echte Langläufer in ein externes Arbeitssystem
(Celery, RQ).
Die TaskGroup wirft eine ExceptionGroup; except* greift ein Mitglied
heraus. Ein raise innerhalb von except* wird neu verpackt — Kennzeichen
setzen und nach dem Block werfen.
Modul: Nebenläufigkeit härten
Server-Sent Events
| Feld | Bedeutung |
|---|---|
data | die Nutzlast |
event | Name des Typs |
id | laufende Nummer, Browser merkt sie sich |
retry | Wartezeit in ms bis zum Wiederverbinden |
Frame ist data: … plus Leerzeile, Medientyp text/event-stream. EventSource
verbindet von selbst neu und schickt Last-Event-ID mit — die Wiederaufnahme
müssen Sie liefern. Im Betrieb: Proxy-Pufferung für die Route abschalten,
regelmäßig eine Kommentarzeile (:) als Lebenszeichen senden, über HTTP/2
ausliefern.
Modul: Nebenläufigkeit härten
Geteilter Zustand
| Zustand | im Prozess | besser |
|---|---|---|
| Verbindungspool | richtig so | — |
| Zähler, Kennzahlen | je Arbeiter falsch | externer Speicher |
| Cache | bestenfalls Vorstufe | externer Speicher |
| Rate Limit | wirkungslos | externer Speicher |
Locks synchronisieren Threads, nicht Prozesse. Mit mehreren Arbeitsprozessen hat jeder seine eigene Kopie von Modul, Dictionary und Lock. Prüfliste: Modulvariablen, Klassenattribute, jeder Cache, den mehr als ein Thread berührt. Free-Threading ist seit Python 3.14 offiziell unterstützt (PEP 779), nicht mehr experimentell wie unter 3.13t.
Modul: Nebenläufigkeit härten
Typische Fallen
model_dump()ohnemode="json"liefert Python-Objekte; eindatebleibt ein Datumsobjekt und knallt erst downstream.model_dump()ohneby_aliasschickt snake_case nach außen und bricht still den Vertrag, den die Anfrage einhält.@model_validator(mode="after")ohnereturn selfgibtNonezurück — das Modell ist danach leer, ohne Fehlermeldung.- Alias nur auf dem Create-Modell gesetzt: Die Anfrage spricht camelCase, die Antwort snake_case.
- Fehlendes
expire_on_commit=False: Der Zugriff nach dem Commit scheitert mitMissingGreenlet— weit weg von der Ursache. - Eine Engine je Anfrage statt je Prozess: eigener Pool pro Anfrage, fällt erst unter Last auf.
- Ungeladene Beziehung in Async berühren wirft
MissingGreenlet, statt still nachzuladen. Das ist die gute Nachricht — synchron wäre es nur langsam. - Eine Ladeoption für zwei Ebenen: Jeder Sprung braucht seine eigene.
- Yield-Dependency ohne
raisenach dem Rollback verschluckt den Fehler; der Client bekommt 200. - Injizierte Session im SSE-Generator: Sie ist abgebaut, sobald die Antwort erzeugt ist — der Generator sendet danach weiter. Eigene kurzlebige Session öffnen.
HTTPExceptioninnerhalb vonexcept*wird neu verpackt und erreicht FastAPI nie; der Client sieht 500 statt 409.- Blockierender Code in einer
async def-Dependency legt den Event Loop lahm. Für reine Konstruktiondefnehmen — FastAPI schiebt sie in den Threadpool.