ドキュメント価格ブログ
お問い合わせ
ログイン無料で新規登録
Back to Tech Articles

Asynchrone Datenbankoperationen in FastAPI mit SQLModel und Tortoise ORM

7月 28, 2025
# Misc

Ethan Miller

Product Engineer · Leapcell

この記事はEnglish、日本語、한국어でも利用できます
Cover of "Asynchrone Datenbankoperationen in FastAPI mit SQLModel und Tortoise ORM"

Einleitung

In der sich schnell entwickelnden Landschaft der Backend-Entwicklung ist der Aufbau leistungsstarker, skalierbarer Webdienste von größter Bedeutung. Asynchrone Programmierung hat sich als Eckpfeiler für die Erreichung solcher Ziele erwiesen und ermöglicht es Anwendungen, zahlreiche gleichzeitige Anfragen zu bearbeiten, ohne den Hauptthread zu blockieren. FastAPI hat sich mit seiner inhärenten Unterstützung für asynchrone Operationen und seinem intuitiven Design schnell zum bevorzugten Framework für den Aufbau moderner APIs entwickelt. Die Vorteile von asynchronen Web-Frameworks können jedoch erheblich beeinträchtigt werden, wenn Datenbankinteraktionen synchron bleiben. Dieser Engpass ist genau das, was asynchrone ORMs und Datenbankbibliotheken lösen sollen. Dieser Artikel untersucht, wie zwei prominente asynchrone Datenbankwerkzeuge – SQLModel und Tortoise ORM – in FastAPI integriert werden, um wirklich nicht-blockierende Datenbankoperationen zu ermöglichen und so die Gesamteffizienz und Reaktionsfähigkeit Ihrer Backend-Dienste zu verbessern.

Kernkonzepte und Prinzipien

Bevor wir uns mit den Implementierungsdetails befassen, lassen Sie uns einige grundlegende Begriffe klären, die für das Verständnis asynchroner Datenbankoperationen in FastAPI von entscheidender Bedeutung sind.

  • Asynchrone Programmierung: Ein Programmierparadigma, das es einem Programm ermöglicht, eine lang dauernde Aufgabe auszuführen, ohne die gesamte Anwendung einzufrieren. In Python wird dies hauptsächlich mit den Schlüsselwörtern async/await und der asyncio-Bibliothek erreicht, die kooperatives Multitasking ermöglicht.
  • ORM (Object-Relational Mapping): Eine Technik, die es Entwicklern ermöglicht, mit einer Datenbank über ein objektorientiertes Paradigma zu interagieren. Anstatt rohe SQL-Abfragen zu schreiben, ermöglichen ORMs die Manipulation von Datenbankdatensätzen als Objekte in Ihrer bevorzugten Programmiersprache. Diese Abstraktion vereinfacht Datenbankoperationen, verbessert die Lesbarkeit des Codes und erhöht oft die Sicherheit, indem SQL-Injection verhindert wird.
  • FastAPI: Ein modernes, schnelles (hochperformantes) Web-Framework zum Erstellen von APIs mit Python 3.7+ auf der Grundlage von Standard-Python-Typ-Hinweisen. Es ist tief in asyncio integriert, was es ideal für asynchrone Webdienste macht.
  • SQLModel: Eine Python-Bibliothek zur Interaktion mit SQL-Datenbanken, die sowohl "SQL- als auch Python-first" konzipiert wurde. Sie basiert auf Pydantic und SQLAlchemy und zielt darauf ab, eine einfachere, intuitivere Erfahrung für die Definition von Modellen zu bieten, die sowohl als Pydantic-Modelle für die Datenvalidierung als auch als SQLAlchemy-Modelle für die Datenbankinteraktion dienen. Sie unterstützt inhärent asynchrone Operationen über den asynchronen Engine von SQLAlchemy.
  • Tortoise ORM: Ein einfach zu bedienendes asynchrones ORM für Python, das speziell für asyncio und uvloop entwickelt wurde. Es bietet eine einfache API zur Definition von Modellen, zur Durchführung von Abfragen und zur Verwaltung von Datenbankmigrationen, während es gleichzeitig eine asynchrone Natur beibehält.

Das Prinzip hinter der Verwendung asynchroner ORMs mit FastAPI ist einfach: Wenn Ihre FastAPI-Anwendung mit einer Datenbank kommunizieren muss, kann die Anwendung stattdessen auf eine andere Aufgabe umschalten, anstatt synchron auf die Datenbankantwort zu warten. Sobald die Datenbankoperation abgeschlossen ist, kann die Anwendung die Verarbeitung der ursprünglichen Anfrage fortsetzen. Dieses nicht-blockierende Verhalten ist entscheidend für E/A-gebundene Operationen wie Datenbankaufrufe und verhindert, dass der Server unter hoher Last nicht mehr reagiert.

Implementierung asynchroner Datenbankoperationen

Lassen Sie uns untersuchen, wie SQLModel und Tortoise ORM in eine FastAPI-Anwendung für asynchrone Datenbankinteraktionen integriert werden. Zu Demonstrationszwecken verwenden wir ein einfaches "Hero"-Modell.

Verwendung von SQLModel

SQLModel kombiniert Pydantic-Modelle mit SQLAlchemy und bietet eine leistungsstarke und elegante Möglichkeit, Ihre Daten zu definieren.

Einrichtung

Installieren Sie zunächst die notwendigen Pakete:

pip install fastapi "uvicorn[standard]" sqlmodel "psycopg2-binary" # oder asyncpg für async

Wir verwenden PostgreSQL als unsere Datenbank. Für asynchrone Operationen mit PostgreSQL wird oft asyncpg gegenüber psycopg2-binary bevorzugt, aber psycopg2-binary kann mit einem Async-Wrapper verwendet werden. Bleiben wir bei asyncpg für eine native Async-Erfahrung.

pip install fastapi "uvicorn[standard]" sqlmodel asyncpg

Datenbankkonfiguration und Modelldefinition

from typing import Optional, List
from sqlmodel import Field, SQLModel, Session, create_engine
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException, status

# Datenbank-URL, nach Bedarf anpassen
DATABASE_URL = "postgresql+asyncpg://user:password@host:port/dbname"

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class HeroCreate(HeroBase):
    pass

class HeroPublic(HeroBase):
    id: int

# Asynchroner Engine
engine = create_engine(DATABASE_URL, echo=True)

async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

# Abhängigkeit zum Abrufen einer Datenbanksitzung
def get_session():
    with Session(engine) as session:
        yield session

# FastAPI-Anwendungs-Setup
@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

FastAPI-Endpunkte

Erstellen wir nun einige grundlegende CRUD-Endpunkte für das Hero-Modell.

from sqlmodel import select

@app.post("/heroes/", response_model=HeroPublic)
async def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    await session.commit()
    await session.refresh(db_hero)
    return db_hero

@app.get("/heroes/", response_model=List[HeroPublic])
async def read_heroes(offset: int = 0, limit: int = Field(default=100, le=100), session: Session = Depends(get_session)):
    heroes = (await session.exec(select(Hero).offset(offset).limit(limit))).all()
    return heroes

@app.get("/heroes/{hero_id}", response_model=HeroPublic)
async def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = (await session.get(Hero, hero_id))
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

@app.put("/heroes/{hero_id}", response_model=HeroPublic)
async def update_hero(*, session: Session = Depends(get_session), hero_id: int, hero: HeroCreate):
    db_hero = (await session.get(Hero, hero_id))
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    
    hero_data = hero.model_dump(exclude_unset=True)
    for key, value in hero_data.items():
        setattr(db_hero, key, value)
    
    session.add(db_hero)
    await session.commit()
    await session.refresh(db_hero)
    return db_hero

@app.delete("/heroes/{hero_id}")
async def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = (await session.get(Hero, hero_id))
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    await session.delete(hero)
    await session.commit()
    return {"ok": True}

Die Schlüsselaspekte hier sind:

  • create_engine mit postgresql+asyncpg für einen asynchronen Treiber.
  • async with engine.begin() und await conn.run_sync() zum Erstellen von Tabellen asynchron.
  • async with Session(engine) as session: zur Verwaltung asynchroner Datenbanksitzungen.
  • await session.exec() und await session.get() für asynchrone Abfragen.
  • await session.commit() und await session.refresh() zum Speichern von Änderungen und Aktualisieren von Objekten.

Verwendung von Tortoise ORM

Tortoise ORM ist von Grund auf für asynchrone Operationen konzipiert und bietet eine traditionellere ORM-Erfahrung mit eigener Abfragesyntax.

Einrichtung

Installieren Sie Tortoise ORM und Ihren gewählten Async-Datenbanktreiber (z. B. asyncpg für PostgreSQL).

pip install fastapi "uvicorn[standard]" tortoise-orm asyncpg

Datenbankkonfiguration und Modelldefinition

from typing import Optional, List
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from tortoise import fields, models
from tortoise.contrib.fastapi import register_tortoise
from tortoise.exceptions import DoesNotExist

# Datenbankkonfiguration
TORTOISE_CONFIG = {
    "connections": {"default": "postgresql://user:password@host:port/dbname"},
    "apps": {
        "models": {
            "models": ["__main__"], # Angenommen, Modelle sind in dieser Datei
            "default_connection": "default",
        }
    }
}

# Tortoise ORM Hero-Modell
class Hero(models.Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255, unique=True, index=True)
    secret_name = fields.CharField(max_length=255)
    age = fields.IntField(null=True, index=True)

    class Meta:
        table = "heroes"

    def __str__(self):
        return self.name

# Pydantic-Modelle für Request/Response-Validierung
class HeroIn(BaseModel):
    name: str
    secret_name: str
    age: Optional[int] = None

class HeroOut(BaseModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None

async def init_db(app: FastAPI):
    register_tortoise(
        app,
        config=TORTOISE_CONFIG,
        generate_schemas=True, # Erstellt Tabellen, falls sie nicht existieren
        add_exception_handlers=True,
    )

app = FastAPI()

# Registrieren von Tortoise ORM während des App-Starts
@app.on_event("startup")
async def startup_event():
    await init_db(app)

FastAPI-Endpunkte

@app.post("/heroes/", response_model=HeroOut)
async def create_hero(hero_in: HeroIn):
    hero = await Hero.create(**hero_in.model_dump())
    return await HeroOut.from_tortoise_orm(hero)

@app.get("/heroes/", response_model=List[HeroOut])
async def get_heroes(offset: int = 0, limit: int = 100):
    heroes = await Hero.all().offset(offset).limit(limit)
    return [await HeroOut.from_tortoise_orm(hero) for hero in heroes]

@app.get("/heroes/{hero_id}", response_model=HeroOut)
async def get_hero(hero_id: int):
    try:
        hero = await Hero.get(id=hero_id)
        return await HeroOut.from_tortoise_orm(hero)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Hero not found")

@app.put("/heroes/{hero_id}", response_model=HeroOut)
async def update_hero(hero_id: int, hero_in: HeroIn):
    try:
        hero = await Hero.get(id=hero_id)
        await hero.update_from_dict(hero_in.model_dump(exclude_unset=True))
        await hero.save()
        return await HeroOut.from_tortoise_orm(hero)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Hero not found")

@app.delete("/heroes/{hero_id}", status_code=204)
async def delete_hero(hero_id: int):
    try:
        hero = await Hero.get(id=hero_id)
        await hero.delete()
        return {"message": "Hero deleted successfully"}
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Hero not found")

Mit Tortoise ORM:

  • register_tortoise verwaltet die Datenbankverbindung und die Schemagenerierung.
  • Modelle erben von models.Model und verwenden fields zur Definition von Attributen.
  • await Hero.create(), await Hero.all(), await Hero.get(), await hero.save(), await hero.delete() sind alles asynchrone Operationen.
  • HeroOut.from_tortoise_orm() ist eine praktische Methode, die von tortoise.contrib.pydantic bereitgestellt wird, um eine ORM-Instanz in ein Pydantic-Modell zu konvertieren.

Anwendungsfälle

Sowohl SQLModel als auch Tortoise ORM eignen sich hervorragend für Szenarien, in denen:

  • Hohe Nebenläufigkeit erwartet wird: Eine große Anzahl von Benutzern oder Anfragen erfordert eine effiziente E/A-Verarbeitung.
  • Microservices-Architekturen: Entkoppelte Dienste profitieren oft von einer schnellen, nicht-blockierenden Kommunikation mit ihren jeweiligen Datenbanken.
  • Echtzeitanwendungen: APIs, die Echtzeitdaten oder Updates bereitstellen, erfordern Datenbankoperationen mit geringer Latenz.
  • Moderne Python-Backends: Die Integration mit Frameworks wie FastAPI nutzt deren asynchrone Fähigkeiten auf natürliche Weise.

Die Stärken von SQLModel liegen in seiner engen Integration mit Pydantic, was es für Projekte auszeichnet, bei denen Datenvalidierung und -serialisierung entscheidend sind und Sie eine einzige Quelle der Wahrheit für Ihre Datenmodelle wünschen. Es basiert auf der robusten Grundlage von SQLAlchemy und bietet fortschrittliche Abfragemöglichkeiten.

Die Stärken von Tortoise ORM umfassen seine einfachere API, die speziell für asyncio entwickelt wurde und oft eine sanftere Lernkurve für Neulinge in asynchronen ORMs bietet. Es ist besonders attraktiv, wenn Sie eine eigenständigere ORM-Lösung mit eigener Syntax für Abfragen bevorzugen.

Fazit

Die Integration asynchroner Datenbankoperationen in Ihre FastAPI-Anwendungen mit ORMs wie SQLModel oder Tortoise ORM ist ein entscheidender Schritt zum Aufbau leistungsfähiger und skalierbarer Webdienste. Beide Werkzeuge bieten robuste, asynchrone Funktionen, die E/A-Engpässe beseitigen und sicherstellen, dass Ihre Anwendung auch unter hoher Last reaktionsfähig bleibt. SQLModel bietet eine einheitliche Modelldefinition mit der Leistung von Pydantic und SQLAlchemy, während Tortoise ORM eine prägnante, asyncio-native Erfahrung bietet. Die Wahl zwischen ihnen hängt oft von den spezifischen Projektanforderungen, der Vertrautheit des Teams und der Präferenz für ihre jeweiligen API-Stile ab, aber beide werden die Datenbankinteraktionseffizienz Ihrer FastAPI-Anwendung erheblich verbessern. Durch die Nutzung asynchroner ORMs können Sie die Fähigkeiten von FastAPI voll ausschöpfen und wirklich nicht-blockierende, leistungsstarke Backends bereitstellen.

Share this article

HackerNews

More Posts from Leapcell

Popular Posts