Einfache Hero API mit FastAPI¶
Lassen Sie uns mit dem Aufbau einer einfachen Web-API für Helden mit FastAPI beginnen. ✨
FastAPI installieren¶
Der erste Schritt ist die Installation von FastAPI.
FastAPI ist das Framework zur Erstellung der Web-API.
Stellen Sie sicher, dass Sie eine virtuelle Umgebung erstellen, diese aktivieren und dann die Pakete installieren, z. B. mit
$ pip install fastapi "uvicorn[standard]"
---> 100%
SQLModel Code - Modelle, Engine¶
Nun beginnen wir mit dem SQLModel-Code.
Wir beginnen mit der einfachsten Version, nur mit Helden (noch keine Teams).
Das ist fast derselbe Code, den wir bisher in früheren Beispielen gesehen haben.
# Code above omitted 👆
from sqlmodel import Field, Session, SQLModel, create_engine, select
# Code here omitted 👈
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)
# 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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
Hier gibt es nur eine Änderung gegenüber dem Code, den wir zuvor verwendet haben: check_same_thread in den connect_args.
Das ist eine Konfiguration, die SQLAlchemy an die Low-Level-Bibliothek weitergibt, die für die Kommunikation mit der Datenbank zuständig ist.
check_same_thread ist standardmäßig auf True gesetzt, um Missbrauch in einigen einfachen Fällen zu verhindern.
Aber hier stellen wir sicher, dass wir nicht dieselbe Session für mehr als einen Request verwenden, und das ist der tatsächlich sicherste Weg, um Probleme zu vermeiden, für die diese Konfiguration gedacht ist.
Und wir müssen sie auch deaktivieren, da in FastAPI jeder Request von mehreren interagierenden Threads bearbeitet werden kann.
Info
Das ist vorerst genug Information, Sie können mehr darüber in den FastAPI-Dokumenten für async und await lesen.
Der Hauptpunkt ist, dass der Code bereits sicher ist, wenn Sie sicherstellen, dass Sie dieselbe Session nicht mit mehr als einem Request teilen.
FastAPI App¶
Der nächste Schritt ist die Erstellung der FastAPI-App.
Wir werden die FastAPI-Klasse aus fastapi importieren.
Und dann ein app-Objekt erstellen, das eine Instanz dieser FastAPI-Klasse ist.
from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select
# Code here omitted 👈
app = FastAPI()
# 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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
Datenbank und Tabellen beim startup erstellen¶
Wir möchten sicherstellen, dass die Funktion create_db_and_tables aufgerufen wird, sobald die App gestartet wird. Zum Erstellen der Datenbank und Tabellen.
Dies sollte nur einmal beim Start aufgerufen werden, nicht vor jedem Request, daher packen wir es in die Funktion, die das "startup"-Ereignis behandelt.
# Code above omitted 👆
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
# 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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
Heroes Path Operation erstellen¶
Info
Wenn Sie eine Auffrischung benötigen, was eine Path Operation ist (ein Endpunkt mit einer spezifischen HTTP-Operation) und wie man damit in FastAPI arbeitet, schauen Sie sich die FastAPI First Steps docs an.
Lassen Sie uns den Path Operation-Code zum Erstellen eines neuen Helden erstellen.
Er wird aufgerufen, wenn ein Benutzer eine Anfrage mit einer POST-Operation an den Pfad /heroes/ sendet.
# Code above omitted 👆
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/")
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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
Info
Wenn Sie eine Auffrischung einiger dieser Konzepte benötigen, schauen Sie sich die FastAPI-Dokumentation an
Der SQLModel Vorteil¶
Hier glänzt die Tatsache, dass unsere SQLModel-Klassenmodelle gleichzeitig SQLAlchemy-Modelle und Pydantic-Modelle sind. ✨
Hier verwenden wir dasselbe Klassenmodell, um den Request Body zu definieren, der von unserer API empfangen wird.
Da FastAPI auf Pydantic basiert, verwendet es dasselbe Modell (den Pydantic-Teil), um automatische Datenvalidierung und Konvertierung von der JSON-Anfrage in ein Objekt zu realisieren, das eine tatsächliche Instanz der Hero-Klasse ist.
Und da dieses SQLModel-Objekt nicht nur eine Pydantic-Modellinstanz, sondern auch eine SQLAlchemy-Modellinstanz ist, können wir es direkt in einer Session verwenden, um die Zeile in der Datenbank zu erstellen.
So können wir intuitive Standard-Python-Typ-Annotationen verwenden und müssen nicht viel Code für die Datenbankmodelle und die API-Datenmodelle duplizieren. 🎉
Tipp
Wir werden dies später weiter verbessern, aber vorerst zeigt es bereits die Stärke von SQLModel-Klassen, die gleichzeitig SQLAlchemy-Modelle und Pydantic-Modelle sind.
Heroes Path Operation lesen¶
Fügen wir nun eine weitere Path Operation hinzu, um alle Helden zu lesen.
# Code above omitted 👆
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
with Session(engine) as session:
session.add(hero)
session.commit()
session.refresh(hero)
return hero
@app.get("/heroes/")
def read_heroes():
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
return heroes
Das ist ziemlich einfach.
Wenn ein Client eine Anfrage an den Pfad /heroes/ mit einer GET HTTP-Operation sendet, führen wir diese Funktion aus, die die Helden aus der Datenbank abruft und sie zurückgibt.
Eine Session pro Request¶
Erinnern Sie sich, dass wir eine SQLModel-Session für jede Gruppe von Operationen verwenden sollten und wenn wir andere, nicht zusammenhängende Operationen benötigen, sollten wir eine andere Session verwenden?
Hier ist es viel offensichtlicher.
Wir sollten in den meisten Fällen normalerweise eine Session pro Request haben.
In einigen isolierten Fällen möchten wir innerhalb weitere Sessions erstellen, also mehr als eine Session pro Request.
Aber wir würden niemals dieselbe Session über verschiedene Requests hinweg teilen wollen.
In diesem einfachen Beispiel erstellen wir die neuen Sessions manuell in den Path Operation Funktionen.
In zukünftigen Beispielen werden wir eine FastAPI Dependency verwenden, um die Session zu erhalten, sie mit anderen Abhängigkeiten teilen zu können und sie beim Testen ersetzen zu können. 🤓
Den FastAPI Server im Entwicklungsmodus ausführen¶
Nun sind wir bereit, die FastAPI-Anwendung auszuführen.
Speichern Sie all diesen Code in einer Datei namens main.py.
Führen Sie sie dann mit dem fastapi CLI im Entwicklungsmodus aus.
$ fastapi dev main.py
<span style="color: green;">INFO</span>: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
Info
Der Befehl fastapi verwendet im Hintergrund Uvicorn.
Wenn Sie fastapi dev verwenden, startet es Uvicorn mit der Option, automatisch neu zu laden, wenn Sie eine Änderung am Code vornehmen. So können Sie schneller entwickeln. 🤓
Den FastAPI Server im Produktionsmodus ausführen¶
Der Entwicklungsmodus sollte nicht in der Produktion verwendet werden, da er standardmäßig mit automatischem Neuladen mehr Ressourcen verbraucht als nötig und fehleranfälliger wäre usw.
Verwenden Sie für die Produktion stattdessen fastapi run anstelle von fastapi dev.
$ fastapi run main.py
<span style="color: green;">INFO</span>: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Die API-Dokumentations-UI prüfen¶
Jetzt können Sie diese URL in Ihrem Browser unter http://127.0.0.1:8000 aufrufen. Wir haben keine Path Operation für den Root-Pfad / erstellt, daher zeigt diese URL allein einen "Not Found"-Fehler an... dieser "Not Found"-Fehler wird von Ihrer FastAPI-Anwendung erzeugt.
Aber Sie können die automatisch generierte interaktive API-Dokumentation unter dem Pfad /docs aufrufen: http://127.0.0.1:8000/docs. ✨
Sie werden sehen, dass diese automatische API-Docs-UI die oben definierten Pfade mit ihren Operationen enthält und dass sie bereits die Struktur der Daten kennt, die die Path Operations empfangen werden.
Mit der API spielen¶
Sie können tatsächlich auf die Schaltfläche Try it out klicken und einige Anfragen senden, um einige Helden mit der Path Operation Create Hero zu erstellen.
Und dann können Sie sie mit der Path Operation Read Heroes zurückholen.
Die Datenbank prüfen¶
Jetzt können Sie das Serverprogramm beenden, indem Sie zum Terminal zurückkehren und Ctrl+C drücken.
Und dann können Sie DB Browser for SQLite öffnen und die Datenbank prüfen, um die Daten zu erkunden und zu bestätigen, dass die Helden tatsächlich gespeichert wurden. 🎉
Zusammenfassung¶
Gut gemacht! Dies ist bereits eine FastAPI-Web-API-Anwendung zur Interaktion mit der Helden-Datenbank. 🎉
Es gibt mehrere Dinge, die wir verbessern und erweitern können. Zum Beispiel möchten wir, dass die Datenbank die ID jedes neuen Helden bestimmt, wir möchten nicht zulassen, dass ein Benutzer sie sendet.
All diese Verbesserungen werden wir in den nächsten Kapiteln vornehmen. 🚀