FastAPI Response Model mit SQLModel¶
Jetzt zeige ich Ihnen, wie Sie das response_model von FastAPI mit SQLModel verwenden.
Interaktive API-Dokumentation¶
Bis jetzt weiß die API-Dokumentation mit dem von uns verwendeten Code, welche Daten die Clients senden müssen.
Diese interaktive Dokumentations-UI wird von Swagger UI betrieben. Swagger UI liest einen großen JSON-Inhalt, der die API mit allen Datenschemata (Datenformen) gemäß dem Standard OpenAPI definiert und sie in dieser schönen UI anzeigt.
FastAPI generiert diese OpenAPI automatisch, damit Swagger UI sie lesen kann.
Und es generiert sie basierend auf dem Code, den Sie schreiben, unter Verwendung der Pydantic-Modelle (in diesem Fall SQLModel-Modelle) und Typ-Annotationen, um die Schemata der Daten zu kennen, die die API verarbeitet.
Antwortdaten¶
Aber bis jetzt kennt die API-Dokumentations-UI das Schema der *Antworten*, die unsere App zurücksendet, nicht.
Sie sehen, dass es eine mögliche "Erfolgreiche Antwort" mit dem Code 200 gibt, aber wir haben keine Ahnung, wie die Antwortdaten aussehen würden.
Im Moment teilen wir FastAPI nur die Daten mit, die wir empfangen möchten, aber wir teilen ihm noch nicht die Daten mit, die wir zurücksenden möchten.
Machen wir das jetzt. 🤓
Verwenden von response_model¶
Wir können response_model verwenden, um FastAPI das Schema der Daten mitzuteilen, die wir zurücksenden möchten.
Zum Beispiel können wir dieselbe SQLModel-Klasse Hero übergeben (da sie auch ein Pydantic-Modell ist).
# Code above omitted 👆
@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
# Code below omitted 👇
👀 Vorschau der vollständigen Datei
from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
🤓 Andere Versionen und Varianten
from typing import Optional
from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
from typing import List, Optional
from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/", response_model=List[Hero])
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
Liste von Helden in response_model¶
Wir können auch andere Typ-Annotationen verwenden, genauso wie wir es mit Pydantic-Feldern tun können. Zum Beispiel können wir eine Liste von Heros übergeben.
Dazu deklarieren wir response_model mit list[Hero].
# Code above omitted 👆
@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
👀 Vorschau der vollständigen Datei
from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: int | None = Field(default=None, index=True)
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
🤓 Andere Versionen und Varianten
from typing import Optional
from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
from typing import List, Optional
from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/", response_model=List[Hero])
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
FastAPI und Response Model¶
FastAPI führt mit diesem response_model eine Datenvalidierung und -filterung der Antwort durch.
Dies funktioniert also wie ein Vertrag zwischen unserer Anwendung und dem Client.
Mehr dazu erfahren Sie in der FastAPI-Dokumentation zu response_model.
Neue API-Dokumentations-UI¶
Nun können wir zurück zur Dokumentations-UI gehen und sehen, dass sie jetzt das Schema der Antwort anzeigen, die wir erhalten werden.
Die Clients wissen, welche Daten sie erwarten sollten.
Automatische Clients¶
Der sichtbarste Vorteil der Verwendung von response_model ist, dass es in der API-Dokumentations-UI angezeigt wird.
Es gibt aber auch andere Vorteile, wie z.B. dass FastAPI mit diesem Modell eine automatische Datenvalidierung und -filterung der Antwortdaten durchführt.
Darüber hinaus können viele Tools davon profitieren, da die Schemata standardkonform definiert sind.
Zum Beispiel Client-Generatoren, die automatisch den notwendigen Code zum Kommunizieren mit Ihrer API in vielen Sprachen erstellen können.
Info
Wenn Sie neugierig auf die Standards sind: FastAPI generiert OpenAPI, das intern JSON Schema verwendet.
All das können Sie in der FastAPI-Dokumentation - Erste Schritte nachlesen.
Zusammenfassung¶
Verwenden Sie response_model, um FastAPI das Schema der Daten mitzuteilen, die Sie zurücksenden möchten, und erstellen Sie großartige Daten-APIs. 😎