Muster und Praktiken für die Verwendung von SQLAlchemy 2.0 mit FastAPI

Während Django und Flask für viele Python-Ingenieure nach wie vor die erste Wahl sind, wurde FastAPI bereits als unbestreitbar zuverlässige Wahl anerkannt. Es handelt sich um ein äußerst flexibles, gut optimiertes und strukturiertes Framework, das dem Entwickler endlose Möglichkeiten zum Erstellen von Backend-Anwendungen bietet.

Die Arbeit mit Datenbanken ist ein wesentlicher Aspekt der meisten Backend-Anwendungen. Daher spielt der ORM eine entscheidende Rolle im Backend-Code. Im Gegensatz zu Django verfügt FastAPI jedoch nicht über ein integriertes ORM. Es liegt ausschließlich in der Verantwortung des Entwicklers, eine geeignete Bibliothek auszuwählen und in die Codebasis zu integrieren.

Python-Ingenieure halten SQLAlchemy allgemein für das beliebteste verfügbare ORM. Es handelt sich um eine legendäre Bibliothek, die seit 2006 verwendet wird und von Tausenden von Projekten übernommen wurde. Im Jahr 2023 erhielt es ein großes Update auf Version 2.0. Ähnlich wie FastAPI bietet SQLAlchemy Entwicklern leistungsstarke Funktionen und Dienstprogramme, ohne sie zu einer bestimmten Verwendung zu zwingen. Im Wesentlichen handelt es sich um ein vielseitiges Toolkit, das es Entwicklern ermöglicht, es so zu verwenden, wie sie es für richtig halten.

FastAPI und SQLAlchemy sind eine himmlische Kombination. Dabei handelt es sich um zuverlässige, leistungsstarke und moderne Technologien, die die Erstellung leistungsstarker und einzigartiger Anwendungen ermöglichen. In diesem Artikel wird die Erstellung einer FastAPI-Backend-Anwendung untersucht, die SQLAlchemy 2.0 als ORM verwendet. Der Inhalt umfasst:

• Erstellen von Modellen mit Mapped und mapped_column
• Definieren eines abstrakten Modells
• Umgang mit Datenbanksitzungen
• unter Verwendung des ORM
• Erstellen einer gemeinsamen Repository-Klasse für alle Modelle
• Vorbereiten eines Testaufbaus und Hinzufügen von Tests

Anschließend können Sie die FastAPI-Anwendung problemlos mit SQLAlchemy ORM kombinieren. Darüber hinaus machen Sie sich mit Best Practices und Mustern zur Erstellung gut strukturierter, robuster und leistungsstarker Anwendungen vertraut.

Voraussetzungen

Die im Artikel enthaltenen Codebeispiele stammen aus dem Alchemist- Projekt, bei dem es sich um eine grundlegende API zum Erstellen und Lesen von Zutaten- und Trankobjekten handelt. Der Schwerpunkt des Artikels liegt auf der Untersuchung der Kombination von FastAPI und SQLAlchemy. Andere Themen werden nicht behandelt, z. B.:

• Konfigurieren des Docker-Setups
• Starten des Uvicorn-Servers
• Flusen einrichten

Wenn Sie an diesen Themen interessiert sind, können Sie sie selbst erkunden, indem Sie die Codebasis untersuchen. Um auf das Code-Repository des Alchemist-Projekts zuzugreifen, folgen Sie bitte diesem Link hier . Darüber hinaus finden Sie unten die Dateistruktur des Projekts:

 alchemist ├─ alchemist │ ├─ api │ │ ├─ v1 │ │ │ ├─ __init__.py │ │ │ └─ routes.py │ │ ├─ v2 │ │ │ ├─ __init__.py │ │ │ ├─ dependencies.py │ │ │ └─ routes.py │ │ ├─ __init__.py │ │ └─ models.py │ ├─ database │ │ ├─ __init__.py │ │ ├─ models.py │ │ ├─ repository.py │ │ └─ session.py │ ├─ __init__.py │ ├─ app.py │ └─ config.py ├─ requirements │ ├─ base.txt │ └─ dev.txt ├─ scripts │ ├─ create_test_db.sh │ ├─ migrate.py │ └─ run.sh ├─ tests │ ├─ conftest.py │ └─ test_api.py ├─ .env ├─ .gitignore ├─ .pre-commit-config.yaml ├─ Dockerfile ├─ Makefile ├─ README.md ├─ docker-compose.yaml ├─ example.env └─ pyproject.toml

Obwohl der Baum groß erscheinen mag, sind einige Inhalte für den Hauptpunkt dieses Artikels nicht relevant. Darüber hinaus kann der Code in bestimmten Bereichen einfacher erscheinen als erforderlich. Dem Projekt fehlen beispielsweise:

• Produktionsphase in der Docker-Datei
• Destillierkolben-Setup für Migrationen
• Unterverzeichnisse für Tests

Dies geschah bewusst, um die Komplexität zu reduzieren und unnötigen Overhead zu vermeiden. Es ist jedoch wichtig, diese Faktoren im Auge zu behalten, wenn es sich um ein produktionsreiferes Projekt handelt.

API-Anforderungen

Wenn Sie mit der Entwicklung einer App beginnen, ist es wichtig, die Modelle zu berücksichtigen, die Ihre App verwenden wird. Diese Modelle stellen die Objekte und Entitäten dar, mit denen Ihre App arbeitet, und werden in der API verfügbar gemacht. Im Fall der Alchemist-App gibt es zwei Entitäten: Zutaten und Tränke. Die API sollte das Erstellen und Abrufen dieser Entitäten ermöglichen. Die Datei alchemist/api/models.py enthält die Modelle, die in der API verwendet werden:

 import uuid from pydantic import BaseModel, Field class Ingredient(BaseModel): """Ingredient model.""" pk: uuid.UUID name: str class Config: orm_mode = True class IngredientPayload(BaseModel): """Ingredient payload model.""" name: str = Field(min_length=1, max_length=127) class Potion(BaseModel): """Potion model.""" pk: uuid.UUID name: str ingredients: list[Ingredient] class Config: orm_mode = True class PotionPayload(BaseModel): """Potion payload model.""" name: str = Field(min_length=1, max_length=127) ingredients: list[uuid.UUID] = Field(min_items=1)

Die API gibt Ingredient und Potion zurück. Wenn Sie orm_mode in der Konfiguration auf True setzen, wird es in Zukunft einfacher, mit den SQLAlchemy-Objekten zu arbeiten. Payload werden zum Erstellen neuer Objekte verwendet.

Durch den Einsatz von Pydantic werden die Klassen detaillierter und klarer in ihren Rollen und Funktionen. Jetzt ist es an der Zeit, die Datenbankmodelle zu erstellen.

Modelle deklarieren

Ein Modell ist im Wesentlichen eine Darstellung von etwas. Im Kontext von APIs stellen Modelle dar, was das Backend im Anfragetext erwartet und was es in den Antwortdaten zurückgibt. Datenbankmodelle hingegen sind komplexer und repräsentieren die in der Datenbank gespeicherten Datenstrukturen und die Beziehungstypen zwischen ihnen.

Die Datei alchemist/database/models.py enthält Modelle für Zutaten- und Trankobjekte:

 import uuid from sqlalchemy import Column, ForeignKey, Table, orm from sqlalchemy.dialects.postgresql import UUID class Base(orm.DeclarativeBase): """Base database model.""" pk: orm.Mapped[uuid.UUID] = orm.mapped_column( primary_key=True, default=uuid.uuid4, ) potion_ingredient_association = Table( "potion_ingredient", Base.metadata, Column("potion_id", UUID(as_uuid=True), ForeignKey("potion.pk")), Column("ingredient_id", UUID(as_uuid=True), ForeignKey("ingredient.pk")), ) class Ingredient(Base): """Ingredient database model.""" __tablename__ = "ingredient" name: orm.Mapped[str] class Potion(Base): """Potion database model.""" __tablename__ = "potion" name: orm.Mapped[str] ingredients: orm.Mapped[list["Ingredient"]] = orm.relationship( secondary=potion_ingredient_association, backref="potions", lazy="selectin", )

Jedes Modell in SQLAlchemy beginnt mit der DeclarativeBase Klasse. Durch die Vererbung davon können Datenbankmodelle erstellt werden, die mit Python-Typprüfern kompatibel sind.

Es empfiehlt sich auch, ein abstraktes Modell – in diesem Fall Base – zu erstellen, das die in allen Modellen erforderlichen Felder enthält. Zu diesen Feldern gehört der Primärschlüssel, der eine eindeutige Kennung jedes Objekts darstellt. Das abstrakte Modell speichert häufig auch die Erstellungs- und Aktualisierungsdaten eines Objekts, die automatisch festgelegt werden, wenn ein Objekt erstellt oder aktualisiert wird. Das Base wird jedoch einfach gehalten.

Beim Ingredient Modell gibt das Attribut __tablename__ den Namen der Datenbanktabelle an, während das name die neue SQLAlchemy-Syntax verwendet, sodass Modellfelder mit Typanmerkungen deklariert werden können. Dieser prägnante und moderne Ansatz ist sowohl leistungsstark als auch vorteilhaft für Typprüfer und IDEs, da er das name als Zeichenfolge erkennt.

Im Potion wird es komplexer. Es enthält auch die Attribute __tablename__ und name , speichert aber darüber hinaus die Beziehung zu Zutaten. Die Verwendung von Mapped[list["Ingredient"]] zeigt an, dass der Trank mehrere Zutaten enthalten kann und in diesem Fall die Beziehung viele zu viele (M2M) ist. Das bedeutet, dass eine einzelne Zutat mehreren Tränken zugeordnet werden kann.

M2M erfordert eine zusätzliche Konfiguration, die in der Regel die Erstellung einer Zuordnungstabelle umfasst, in der die Verbindungen zwischen den beiden Entitäten gespeichert werden. In diesem Fall speichert das potion_ingredient_association Objekt nur die Kennungen der Zutat und des Tranks, es könnte aber auch zusätzliche Attribute enthalten, beispielsweise die Menge einer bestimmten Zutat, die für den Trank benötigt wird.

Die relationship konfiguriert die Beziehung zwischen dem Trank und seinen Zutaten. Das Argument lazy gibt an, wie verwandte Elemente geladen werden sollen. Mit anderen Worten: Was soll SQLAlchemy mit den zugehörigen Zutaten tun, wenn Sie einen Trank holen? Die Einstellung „ selectin bedeutet, dass Zutaten mit dem Trank geladen werden, sodass keine zusätzlichen Abfragen im Code erforderlich sind.

Bei der Arbeit mit einem ORM ist die Erstellung gut gestalteter Modelle von entscheidender Bedeutung. Sobald dies erledigt ist, besteht der nächste Schritt darin, die Verbindung mit der Datenbank herzustellen.

Sitzungshandler

Beim Arbeiten mit einer Datenbank, insbesondere bei der Verwendung von SQLAlchemy, ist es wichtig, die folgenden Konzepte zu verstehen:

• Dialekt
• Motor
• Verbindung
• Verbindungspool
• Sitzung

Von all diesen Begriffen ist der Motor der wichtigste. Laut der SQLAlchemy-Dokumentation ist das Engine-Objekt für die Verbindung von Pool und Dialect verantwortlich, um die Konnektivität und das Verhalten der Datenbank zu erleichtern. Einfacher ausgedrückt ist das Engine-Objekt die Quelle der Datenbankverbindung, während die Verbindung High-Level-Funktionalitäten wie das Ausführen von SQL-Anweisungen, das Verwalten von Transaktionen und das Abrufen von Ergebnissen aus der Datenbank bereitstellt.

Eine Sitzung ist eine Arbeitseinheit, die zusammengehörige Vorgänge innerhalb einer einzelnen Transaktion gruppiert. Es handelt sich um eine Abstraktion der zugrunde liegenden Datenbankverbindungen und verwaltet Verbindungen und Transaktionsverhalten effizient.

Dialect ist eine Komponente, die Unterstützung für ein bestimmtes Datenbank-Backend bietet. Es fungiert als Vermittler zwischen SQLAlchemy und der Datenbank und kümmert sich um die Details der Kommunikation. Das Alchemist-Projekt verwendet Postgres als Datenbank, daher muss der Dialekt mit diesem speziellen Datenbanktyp kompatibel sein.

Das letzte Fragezeichen ist der Verbindungspool . Im Kontext von SQLAlchemy ist ein Verbindungspool ein Mechanismus, der eine Sammlung von Datenbankverbindungen verwaltet. Es soll die Leistung und Effizienz von Datenbankoperationen verbessern, indem bestehende Verbindungen wiederverwendet werden, anstatt für jede Anfrage neue zu erstellen. Durch die Wiederverwendung von Verbindungen reduziert der Verbindungspool den Aufwand für den Aufbau neuer Verbindungen und deren Abbau, was zu einer verbesserten Leistung führt.

Mit diesem Wissen können Sie nun einen Blick auf die Datei alchemist/database/session.py werfen, die eine Funktion enthält, die als Abhängigkeit für die Verbindung zur Datenbank verwendet wird:

 from collections.abc import AsyncGenerator from sqlalchemy import exc from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.config import settings async def get_db_session() -> AsyncGenerator[AsyncSession, None]: engine = create_async_engine(settings.DATABASE_URL) factory = async_sessionmaker(engine) async with factory() as session: try: yield session await session.commit() except exc.SQLAlchemyError as error: await session.rollback() raise

Das erste wichtige Detail, das zu beachten ist, ist, dass die Funktion get_db_session eine Generatorfunktion ist. Dies liegt daran, dass das FastAPI-Abhängigkeitssystem Generatoren unterstützt. Daher kann diese Funktion sowohl erfolgreiche als auch fehlgeschlagene Szenarios verarbeiten.

Die ersten beiden Zeilen der Funktion get_db_session erstellen eine Datenbank-Engine und eine Sitzung. Das Sitzungsobjekt kann jedoch auch als Kontextmanager verwendet werden. Dadurch haben Sie mehr Kontrolle über potenzielle Ausnahmen und erfolgreiche Ergebnisse.

Obwohl SQLAlchemy das Schließen von Verbindungen übernimmt, empfiehlt es sich, explizit zu deklarieren, wie mit der Verbindung umgegangen werden soll, nachdem sie hergestellt wurde. In der Funktion get_db_session wird die Sitzung festgeschrieben, wenn alles gut läuft, und ein Rollback durchgeführt, wenn eine Ausnahme ausgelöst wird.

Es ist wichtig zu beachten, dass dieser Code auf der Asyncio-Erweiterung basiert. Diese Funktion von SQLAlchemy ermöglicht der App die asynchrone Interaktion mit der Datenbank. Das bedeutet, dass Anfragen an die Datenbank andere API-Anfragen nicht blockieren, was die App wesentlich effizienter macht.

Sobald die Modelle und die Verbindung eingerichtet sind, besteht der nächste Schritt darin, sicherzustellen, dass die Modelle zur Datenbank hinzugefügt werden.

Schnelle Migrationen

SQLAlchemy-Modelle repräsentieren die Strukturen einer Datenbank. Das bloße Erstellen führt jedoch nicht zu unmittelbaren Änderungen an der Datenbank. Um Änderungen vorzunehmen, müssen Sie diese zunächst anwenden . Dies erfolgt in der Regel mithilfe einer Migrationsbibliothek wie Alembic, die jedes Modell verfolgt und die Datenbank entsprechend aktualisiert.

Da in diesem Szenario keine weiteren Änderungen an den Modellen geplant sind, reicht ein einfaches Migrationsskript aus. Unten finden Sie einen Beispielcode aus der Datei scripts/migrate.py .

 import asyncio import logging from sqlalchemy.ext.asyncio import create_async_engine from alchemist.config import settings from alchemist.database.models import Base logger = logging.getLogger() async def migrate_tables() -> None: logger.info("Starting to migrate") engine = create_async_engine(settings.DATABASE_URL) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("Done migrating") if __name__ == "__main__": asyncio.run(migrate_tables())

Vereinfacht ausgedrückt liest die Funktion migrate_tables die Struktur von Modellen und erstellt sie mithilfe der SQLAlchemy-Engine in der Datenbank neu. Um dieses Skript auszuführen, verwenden Sie den Befehl python scripts/migrate.py .

Die Modelle sind nun sowohl im Code als auch in der Datenbank vorhanden und get_db_session kann Interaktionen mit der Datenbank erleichtern. Sie können jetzt mit der Arbeit an der API-Logik beginnen.

API mit dem ORM

Wie bereits erwähnt, soll die API für Zutaten und Tränke drei Vorgänge unterstützen:

• Objekte erstellen
• Objekte auflisten
• Abrufen von Objekten nach ID

Dank der vorherigen Vorbereitungen können alle diese Funktionen bereits mit SQLAlchemy als ORM und FastAPI als Webframework implementiert werden. Sehen Sie sich zunächst die Zutaten-API an, die sich in der Datei alchemist/api/v1/routes.py befindet:

 import uuid from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.api import models from alchemist.database import models as db_models from alchemist.database.session import get_db_session router = APIRouter(prefix="/v1", tags=["v1"]) @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = db_models.Ingredient(**data.dict()) session.add(ingredient) await session.commit() await session.refresh(ingredient) return models.Ingredient.from_orm(ingredient) @router.get("/ingredients", status_code=status.HTTP_200_OK) async def get_ingredients( session: AsyncSession = Depends(get_db_session), ) -> list[models.Ingredient]: ingredients = await session.scalars(select(db_models.Ingredient)) return [models.Ingredient.from_orm(ingredient) for ingredient in ingredients] @router.get("/ingredients/{pk}", status_code=status.HTTP_200_OK) async def get_ingredient( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = await session.get(db_models.Ingredient, pk) if ingredient is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ingredient does not exist", ) return models.Ingredient.from_orm(ingredient)

Unter der /ingredients API stehen drei Routen zur Verfügung. Der POST-Endpunkt übernimmt eine Inhaltsstoffnutzlast als Objekt aus einem zuvor erstellten Modell und einer Datenbanksitzung. Die Generatorfunktion get_db_session initialisiert die Sitzung und ermöglicht Datenbankinteraktionen.

Im eigentlichen Funktionskörper finden fünf Schritte statt:

1. Aus der eingehenden Nutzlast wird ein Zutatenobjekt erstellt.
2. Die add Methode des Sitzungsobjekts fügt das Inhaltsstoffobjekt dem Sitzungsverfolgungssystem hinzu und markiert es als zum Einfügen in die Datenbank ausstehend.
3. Die Sitzung ist festgeschrieben.
4. Das Inhaltsstoffobjekt wird aktualisiert, um sicherzustellen, dass seine Attribute mit dem Datenbankstatus übereinstimmen.
5. Die Datenbankbestandteilinstanz wird mithilfe der from_orm -Methode in die API-Modellinstanz konvertiert.

Für einen schnellen Test kann ein einfacher Curl gegen die laufende App ausgeführt werden:

 curl -X 'POST' \ 'http://localhost:8000/api/v1/ingredients' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty water"}'

In der Antwort sollte ein Inhaltsstoffobjekt vorhanden sein, dessen ID aus der Datenbank stammt:

 { "pk":"2eb255e9-2172-4c75-9b29-615090e3250d", "name":"Salty water" }

Obwohl die mehreren Abstraktionsebenen von SQLAlchemy für eine einfache API unnötig erscheinen mögen, halten sie die ORM-Details getrennt und tragen zur Effizienz und Skalierbarkeit von SQLAlchemy bei. In Kombination mit Asyncio erbringen die ORM-Funktionen in der API eine außergewöhnlich gute Leistung.

Die verbleibenden beiden Endpunkte sind weniger komplex und weisen Ähnlichkeiten auf. Ein Teil, der eine tiefere Erklärung verdient, ist die Verwendung der scalars innerhalb der Funktion get_ingredients . Beim Abfragen der Datenbank mit SQLAlchemy wird die Methode execute häufig mit einer Abfrage als Argument verwendet. Während die execute zeilenartige Tupel zurückgibt, geben scalars ORM-Entitäten direkt zurück, wodurch der Endpunkt sauberer wird.

Betrachten Sie nun die Tränke-API in derselben Datei:

 @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: data_dict = data.dict() ingredients = await session.scalars( select(db_models.Ingredient).where( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) ) potion = db_models.Potion(**data_dict, ingredients=list(ingredients)) session.add(potion) await session.commit() await session.refresh(potion) return models.Potion.from_orm(potion) @router.get("/potions", status_code=status.HTTP_200_OK) async def get_potions( session: AsyncSession = Depends(get_db_session), ) -> list[models.Potion]: potions = await session.scalars(select(db_models.Potion)) return [models.Potion.from_orm(potion) for potion in potions] @router.get("/potions/{pk}", status_code=status.HTTP_200_OK) async def get_potion( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: potion = await session.get(db_models.Potion, pk) if potion is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Potion does not exist", ) return models.Potion.from_orm(potion)

Die GET-Endpunkte für Tränke sind identisch mit denen für Zutaten. Die POST-Funktion erfordert jedoch zusätzlichen Code. Dies liegt daran, dass beim Erstellen von Tränken mindestens eine Zutaten-ID enthalten sein muss, was bedeutet, dass die Zutaten abgerufen und mit dem neu erstellten Trank verknüpft werden müssen. Um dies zu erreichen, wird erneut die scalars verwendet, dieses Mal jedoch mit einer Abfrage, die die IDs der abgerufenen Zutaten angibt. Der übrige Teil des Trankherstellungsprozesses ist identisch mit dem der Zutaten.

Um den Endpunkt zu testen, kann erneut ein Curl-Befehl ausgeführt werden.

 curl -X 'POST' \ 'http://localhost:8000/api/v1/potions' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty soup", "ingredients": ["0b4f1de5-e780-418d-a74d-927afe8ac954"}'

Es ergibt sich folgende Antwort:

 { "pk": "d4929197-3998-4234-a5f7-917dc4bba421", "name": "Salty soup", "ingredients": [ { "pk": "0b4f1de5-e780-418d-a74d-927afe8ac954", "name": "Salty water" } ] }

Es ist wichtig zu beachten, dass jede Zutat dank des in der Beziehung angegebenen Arguments lazy="selectin" als vollständiges Objekt innerhalb des Tranks dargestellt wird.

Die APIs sind funktionsfähig, aber es gibt ein großes Problem mit dem Code. Während SQLAlchemy Ihnen die Freiheit gibt, nach Belieben mit der Datenbank zu interagieren, bietet es kein übergeordnetes „Manager“-Dienstprogramm ähnlich wie Djangos Model.objects . Daher müssen Sie es selbst erstellen, was im Wesentlichen der Logik entspricht, die in den Zutaten- und Trank-APIs verwendet wird. Wenn Sie diese Logik jedoch direkt in den Endpunkten behalten, ohne sie in einen separaten Bereich zu extrahieren, entsteht am Ende viel duplizierter Code. Darüber hinaus wird es immer schwieriger, Änderungen an den Abfragen oder Modellen zu verwalten.

Das kommende Kapitel stellt das Repository-Muster vor: eine elegante Lösung zum Extrahieren von ORM-Code.

Repository

Das Repository- Muster ermöglicht es, die Details der Arbeit mit der Datenbank zu abstrahieren. Bei der Verwendung von SQLAlchemy, wie im Beispiel des Alchemisten, wäre die Repository-Klasse für die Verwaltung mehrerer Modelle und die Interaktion mit der Datenbanksitzung verantwortlich.

Schauen Sie sich den folgenden Code aus der Datei alchemist/database/repository.py an:

 import uuid from typing import Generic, TypeVar from sqlalchemy import BinaryExpression, select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models Model = TypeVar("Model", bound=models.Base) class DatabaseRepository(Generic[Model]): """Repository for performing database queries.""" def __init__(self, model: type[Model], session: AsyncSession) -> None: self.model = model self.session = session async def create(self, data: dict) -> Model: instance = self.model(**data) self.session.add(instance) await self.session.commit() await self.session.refresh(instance) return instance async def get(self, pk: uuid.UUID) -> Model | None: return await self.session.get(self.model, pk) async def filter( self, *expressions: BinaryExpression, ) -> list[Model]: query = select(self.model) if expressions: query = query.where(*expressions) return list(await self.session.scalars(query))

Die DatabaseRepository Klasse enthält die gesamte Logik, die zuvor in den Endpunkten enthalten war. Der Unterschied besteht darin, dass die spezifische Modellklasse in der __init__ Methode übergeben werden kann, sodass der Code für alle Modelle wiederverwendet werden kann, anstatt ihn an jedem Endpunkt zu duplizieren.

Darüber hinaus verwendet das DatabaseRepository Python-Generika, wobei der generische Typ „ Model an das abstrakte Datenbankmodell gebunden ist. Dadurch kann die Repository-Klasse stärker von der statischen Typprüfung profitieren. Bei Verwendung mit einem bestimmten Modell spiegeln die Rückgabetypen der Repository-Methoden dieses spezifische Modell wider.

Da das Repository die Datenbanksitzung verwenden muss, muss es zusammen mit der Abhängigkeit get_db_session initialisiert werden. Betrachten Sie die neue Abhängigkeit in der Datei alchemist/api/v2/dependencies.py :

 from collections.abc import Callable from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models, repository, session def get_repository(  model: type[models.Base], ) -> Callable[[AsyncSession], repository.DatabaseRepository]:  def func(session: AsyncSession = Depends(session.get_db_session)):    return repository.DatabaseRepository(model, session)  return func

Einfach ausgedrückt ist die Funktion get_repository eine Abhängigkeitsfabrik. Zunächst wird das Datenbankmodell benötigt, mit dem Sie das Repository verwenden möchten. Anschließend wird die Abhängigkeit zurückgegeben, die zum Empfangen der Datenbanksitzung und zum Initialisieren des Repository-Objekts verwendet wird. Um ein besseres Verständnis zu erlangen, sehen Sie sich die neue API in der Datei alchemist/api/v2/routes.py an. Es werden nur die POST-Endpunkte angezeigt, aber es sollte ausreichen, um Ihnen eine klarere Vorstellung davon zu geben, wie der Code verbessert wird:

 from typing import Annotated from fastapi import APIRouter, Depends, status from alchemist.api import models from alchemist.api.v2.dependencies import get_repository from alchemist.database import models as db_models from alchemist.database.repository import DatabaseRepository router = APIRouter(prefix="/v2", tags=["v2"]) IngredientRepository = Annotated[  DatabaseRepository[db_models.Ingredient],  Depends(get_repository(db_models.Ingredient)), ] PotionRepository = Annotated[  DatabaseRepository[db_models.Potion],  Depends(get_repository(db_models.Potion)), ] @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, repository: IngredientRepository, ) -> models.Ingredient: ingredient = await repository.create(data.dict()) return models.Ingredient.from_orm(ingredient) @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, ingredient_repository: IngredientRepository, potion_repository: PotionRepository, ) -> models.Potion: data_dict = data.dict() ingredients = await ingredient_repository.filter( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) potion = await potion_repository.create({**data_dict, "ingredients": ingredients}) return models.Potion.from_orm(potion)

Die erste wichtige Funktion, die es zu beachten gilt, ist die Verwendung von Annotated , einer neuen Art der Arbeit mit FastAPI-Abhängigkeiten. Indem Sie den Rückgabetyp der Abhängigkeit als DatabaseRepository[db_models.Ingredient] angeben und seine Verwendung mit Depends(get_repository(db_models.Ingredient)) deklarieren, können Sie am Ende einfache Typanmerkungen im Endpunkt verwenden: repository: IngredientRepository .

Dank des Repositorys müssen die Endpunkte nicht den gesamten ORM-bezogenen Aufwand speichern. Selbst im komplizierteren Trankfall müssen Sie lediglich zwei Repositories gleichzeitig verwenden.

Sie fragen sich vielleicht, ob die Initialisierung von zwei Repositorys die Sitzung zweimal initialisiert. Die Antwort ist nein. Das FastAPI-Abhängigkeitssystem speichert dieselben Abhängigkeitsaufrufe in einer einzigen Anfrage zwischen. Dies bedeutet, dass die Sitzungsinitialisierung zwischengespeichert wird und beide Repositorys genau dasselbe Sitzungsobjekt verwenden. Ein weiteres großartiges Feature der Kombination von SQLAlchemy und FastAPI.

Die API ist voll funktionsfähig und verfügt über eine wiederverwendbare, leistungsstarke Datenzugriffsschicht. Der nächste Schritt besteht darin, sicherzustellen, dass die Anforderungen erfüllt werden, indem einige End-to-End-Tests geschrieben werden.

Testen

Tests spielen in der Softwareentwicklung eine entscheidende Rolle. Projekte können Unit-, Integrations- und End-to-End-Tests (E2E) enthalten. Während es normalerweise am besten ist, eine große Anzahl aussagekräftiger Unit-Tests durchzuführen, ist es auch gut, mindestens ein paar E2E-Tests zu schreiben, um sicherzustellen, dass der gesamte Workflow ordnungsgemäß funktioniert.

Um einige E2E-Tests für die Alchemist-App zu erstellen, sind zwei zusätzliche Bibliotheken erforderlich:

• pytest, um die Tests tatsächlich zu erstellen und auszuführen
• httpx, um asynchrone Anfragen innerhalb der Tests zu stellen

Sobald diese installiert sind, besteht der nächste Schritt darin, eine separate Testdatenbank einzurichten. Sie möchten nicht, dass Ihre Standarddatenbank verunreinigt oder gelöscht wird. Da alchemist ein Docker-Setup beinhaltet, ist zum Erstellen einer zweiten Datenbank nur ein einfaches Skript erforderlich. Schauen Sie sich den Code aus der Datei scripts/create_test_db.sh an:

 #!/bin/bash psql -U postgres psql -c "CREATE DATABASE test"

Damit das Skript ausgeführt werden kann, muss es als Volume zum Postgres-Container hinzugefügt werden. Dies kann erreicht werden, indem es in den Abschnitt volumes der Datei docker-compose.yaml aufgenommen wird.

Der letzte Schritt der Vorbereitung besteht darin, Pytest-Fixtures in der Datei tests/conftest.py zu erstellen:

 from collections.abc import AsyncGenerator import pytest import pytest_asyncio from fastapi import FastAPI from httpx import AsyncClient from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.app import app from alchemist.config import settings from alchemist.database.models import Base from alchemist.database.session import get_db_session @pytest_asyncio.fixture() async def db_session() -> AsyncGenerator[AsyncSession, None]: """Start a test database session.""" db_name = settings.DATABASE_URL.split("/")[-1] db_url = settings.DATABASE_URL.replace(f"/{db_name}", "/test") engine = create_async_engine(db_url) async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.create_all) session = async_sessionmaker(engine)() yield session await session.close() @pytest.fixture() def test_app(db_session: AsyncSession) -> FastAPI: """Create a test app with overridden dependencies.""" app.dependency_overrides[get_db_session] = lambda: db_session return app @pytest_asyncio.fixture() async def client(test_app: FastAPI) -> AsyncGenerator[AsyncClient, None]: """Create an http client.""" async with AsyncClient(app=test_app, base_url="http://test") as client: yield client

Eine Sache, die in den Tests unbedingt geändert werden muss, ist die Art und Weise, wie die App mit der Datenbank interagiert. Dazu gehört nicht nur das Ändern der Datenbank-URL, sondern auch die Sicherstellung, dass jeder Test isoliert ist, indem mit einer leeren Datenbank begonnen wird.

Das db_session Fixture erreicht beide Ziele. Sein Körper führt die folgenden Schritte aus:

1. Erstellen Sie eine Engine mit einer geänderten Datenbank-URL.
2. Löschen Sie alle vorhandenen Tabellen, um sicherzustellen, dass der Test über eine saubere Datenbank verfügt.
3. Erstellen Sie alle Tabellen innerhalb der Datenbank (gleicher Code wie im Migrationsskript).
4. Erstellen Sie ein Sitzungsobjekt und übergeben Sie es.
5. Schließen Sie die Sitzung manuell, wenn der Test abgeschlossen ist.

Obwohl der letzte Schritt auch als Kontextmanager implementiert werden könnte, funktioniert das manuelle Schließen in diesem Fall einwandfrei.

Die beiden verbleibenden Spielpläne dürften eigentlich selbsterklärend sein:

• test_app ist die FastAPI-Instanz aus der Datei alchemist/app.py , wobei die Abhängigkeit get_db_session durch das Fixture db_session ersetzt wurde
• client ist der httpx AsyncClient , der API-Anfragen an test_app stellt

Nachdem dies alles vorbereitet ist, können endlich die eigentlichen Tests geschrieben werden. Der Einfachheit halber zeigt das folgende Beispiel aus der Datei tests/test_api.py nur einen Test zum Erstellen einer Zutat:

 from fastapi import status class TestIngredientsAPI: """Test cases for the ingredients API.""" async def test_create_ingredient(self, client): response = await client.post("/api/v2/ingredients", json={"name": "Carrot"}) assert response.status_code == status.HTTP_201_CREATED pk = response.json().get("pk") assert pk is not None response = await client.get("/api/v2/ingredients") assert response.status_code == status.HTTP_200_OK assert len(response.json()) == 1 assert response.json()[0]["pk"] == pk

Der Test verwendet ein in einem Fixture erstelltes Client-Objekt, das Anfragen an die FastAPI-Instanz mit überschriebener Abhängigkeit stellt. Dadurch kann der Test mit einer separaten Datenbank interagieren, die nach Abschluss des Tests gelöscht wird. Die Struktur der verbleibenden Testsuite für beide APIs ist weitgehend gleich.

Zusammenfassung

FastAPI und SQLAlchemy sind hervorragende Technologien zur Erstellung moderner und leistungsstarker Backend-Anwendungen. Die Freiheit, Einfachheit und Flexibilität, die sie bieten, machen sie zu einer der besten Optionen für Python-basierte Projekte. Wenn Entwickler Best Practices und Muster befolgen, können sie leistungsstarke, robuste und gut strukturierte Anwendungen erstellen, die Datenbankoperationen und API-Logik problemlos verarbeiten. Dieser Artikel soll Ihnen ein gutes Verständnis dafür vermitteln, wie Sie diese erstaunliche Kombination einrichten und warten.

Quellen

Den Quellcode für das Alchemist-Projekt finden Sie hier: Link .