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 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. FastAPI 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 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. SQLAlchemy 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 und Mapped 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 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.: Alchemist- 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 . Darüber hinaus finden Sie unten die Dateistruktur des Projekts: hier 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 und 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 enthält die Modelle, die in der API verwendet werden: Objekte Entitäten alchemist/api/models.py 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 und zurück. Wenn Sie in der Konfiguration auf setzen, wird es in Zukunft einfacher, mit den SQLAlchemy-Objekten zu arbeiten. werden zum Erstellen neuer Objekte verwendet. Ingredient Potion orm_mode True Payload 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 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 und die zwischen ihnen. Darstellung Datenstrukturen Beziehungstypen Die Datei enthält Modelle für Zutaten- und Trankobjekte: alchemist/database/models.py 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 Klasse. Durch die Vererbung davon können Datenbankmodelle erstellt werden, die mit Python-Typprüfern kompatibel sind. DeclarativeBase Es empfiehlt sich auch, ein Modell – in diesem Fall – 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 wird jedoch einfach gehalten. abstraktes Base Base Beim Modell gibt das Attribut den Namen der Datenbanktabelle an, während das 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 als Zeichenfolge erkennt. Ingredient __tablename__ name name Im wird es komplexer. Es enthält auch die Attribute und , speichert aber darüber hinaus die Beziehung zu Zutaten. Die Verwendung von 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. Potion __tablename__ name Mapped[list["Ingredient"]] 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 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. potion_ingredient_association Die konfiguriert die Beziehung zwischen dem Trank und seinen Zutaten. Das Argument 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 „ bedeutet, dass Zutaten mit dem Trank geladen werden, sodass keine zusätzlichen Abfragen im Code erforderlich sind. relationship lazy selectin 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 der wichtigste. Laut der SQLAlchemy-Dokumentation ist das Engine-Objekt für die Verbindung von und 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 High-Level-Funktionalitäten wie das Ausführen von SQL-Anweisungen, das Verwalten von Transaktionen und das Abrufen von Ergebnissen aus der Datenbank bereitstellt. Motor Pool Dialect Verbindung Eine 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. Sitzung 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. Dialect Das letzte Fragezeichen ist der . 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. Verbindungspool Mit diesem Wissen können Sie nun einen Blick auf die Datei werfen, die eine Funktion enthält, die als Abhängigkeit für die Verbindung zur Datenbank verwendet wird: alchemist/database/session.py 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 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. get_db_session Die ersten beiden Zeilen der Funktion 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. get_db_session 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 wird die Sitzung festgeschrieben, wenn alles gut läuft, und ein Rollback durchgeführt, wenn eine Ausnahme ausgelöst wird. get_db_session 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 . Dies erfolgt in der Regel mithilfe einer Migrationsbibliothek wie Alembic, die jedes Modell verfolgt und die Datenbank entsprechend aktualisiert. anwenden 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 die Struktur von Modellen und erstellt sie mithilfe der SQLAlchemy-Engine in der Datenbank neu. Um dieses Skript auszuführen, verwenden Sie den Befehl . migrate_tables python scripts/migrate.py Die Modelle sind nun sowohl im Code als auch in der Datenbank vorhanden und kann Interaktionen mit der Datenbank erleichtern. Sie können jetzt mit der Arbeit an der API-Logik beginnen. get_db_session 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 befindet: alchemist/api/v1/routes.py 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 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 initialisiert die Sitzung und ermöglicht Datenbankinteraktionen. /ingredients get_db_session Im eigentlichen Funktionskörper finden fünf Schritte statt: Aus der eingehenden Nutzlast wird ein Zutatenobjekt erstellt. Die Methode des Sitzungsobjekts fügt das Inhaltsstoffobjekt dem Sitzungsverfolgungssystem hinzu und markiert es als zum Einfügen in die Datenbank ausstehend. add Die Sitzung ist festgeschrieben. Das Inhaltsstoffobjekt wird aktualisiert, um sicherzustellen, dass seine Attribute mit dem Datenbankstatus übereinstimmen. Die Datenbankbestandteilinstanz wird mithilfe der -Methode in die API-Modellinstanz konvertiert. from_orm 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 innerhalb der Funktion . Beim Abfragen der Datenbank mit SQLAlchemy wird die Methode häufig mit einer Abfrage als Argument verwendet. Während die zeilenartige Tupel zurückgibt, geben ORM-Entitäten direkt zurück, wodurch der Endpunkt sauberer wird. scalars get_ingredients execute execute scalars 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 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. scalars 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 als vollständiges Objekt innerhalb des Tranks dargestellt wird. lazy="selectin" 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 . 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. Model.objects Das kommende Kapitel stellt das Repository-Muster vor: eine elegante Lösung zum Extrahieren von ORM-Code. Repository Das 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. Repository- Schauen Sie sich den folgenden Code aus der Datei an: alchemist/database/repository.py 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 Klasse enthält die gesamte Logik, die zuvor in den Endpunkten enthalten war. Der Unterschied besteht darin, dass die spezifische Modellklasse in der Methode übergeben werden kann, sodass der Code für alle Modelle wiederverwendet werden kann, anstatt ihn an jedem Endpunkt zu duplizieren. DatabaseRepository __init__ Darüber hinaus verwendet das Python-Generika, wobei der generische Typ „ 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. DatabaseRepository Model Da das Repository die Datenbanksitzung verwenden muss, muss es zusammen mit der Abhängigkeit initialisiert werden. Betrachten Sie die neue Abhängigkeit in der Datei : get_db_session 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 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 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: get_repository alchemist/api/v2/routes.py 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 , einer neuen Art der Arbeit mit FastAPI-Abhängigkeiten. Indem Sie den Rückgabetyp der Abhängigkeit als angeben und seine Verwendung mit deklarieren, können Sie am Ende einfache Typanmerkungen im Endpunkt verwenden: . Annotated DatabaseRepository[db_models.Ingredient] Depends(get_repository(db_models.Ingredient)) 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 an: scripts/create_test_db.sh #!/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 der Datei aufgenommen wird. volumes docker-compose.yaml Der letzte Schritt der Vorbereitung besteht darin, Pytest-Fixtures in der Datei zu erstellen: tests/conftest.py 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 Fixture erreicht beide Ziele. Sein Körper führt die folgenden Schritte aus: db_session Erstellen Sie eine Engine mit einer geänderten Datenbank-URL. Löschen Sie alle vorhandenen Tabellen, um sicherzustellen, dass der Test über eine saubere Datenbank verfügt. Erstellen Sie alle Tabellen innerhalb der Datenbank (gleicher Code wie im Migrationsskript). Erstellen Sie ein Sitzungsobjekt und übergeben Sie es. 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: ist die FastAPI-Instanz aus der Datei , wobei die Abhängigkeit durch das Fixture ersetzt wurde test_app alchemist/app.py get_db_session db_session ist der httpx , der API-Anfragen an stellt client AsyncClient test_app Nachdem dies alles vorbereitet ist, können endlich die eigentlichen Tests geschrieben werden. Der Einfachheit halber zeigt das folgende Beispiel aus der Datei nur einen Test zum Erstellen einer Zutat: tests/test_api.py 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 Auch veröffentlicht. hier