FastAPI, Pydantic, Psycopgdie heilige Dreifaltigkeit für Python-Web-APIs

Teil 1: Diskussion

Geben Sie FastAPI ein

Nehmen Sie zunächst den Titel mit einer Prise Salz.

Wenn ich heute mit der Python-Web-API-Entwicklung bei Null anfangen würde, würde ich mir wahrscheinlich LiteStar genauer ansehen, das meiner Meinung nach eine bessere Architektur und eine bessere Projekt-Governance-Struktur zu haben scheint.

Aber wir haben FastAPI und es wird so schnell nirgendwo hingehen. Ich verwende es für viele persönliche und berufliche Projekte und genieße immer noch seine Einfachheit.

Eine Anleitung zu FastAPI-Entwurfsmustern finden Sie auf dieser Seite.

Datenbankdaten abrufen

Obwohl FastAPI im eigentlichen „API“-Teil großartig ist, gab es für mich eine anhaltende Unsicherheit: Wie greife ich am besten auf die Datenbank zu, insbesondere wenn wir auch Geodatentypen verarbeiten müssen.

Lassen Sie uns unsere Optionen überprüfen.

Hinweis 1: Wir sind hier nur an asynchronen Bibliotheken interessiert, da FastAPI ASGI ist.

Hinweis 2: Ich werde nur die Verbindung zu PostgreSQL diskutieren, obwohl Teile der Diskussion immer noch für andere Datenbanken relevant sind.

Einfach zu programmieren | Komplexes Design: ORMs

Verwaltet Ihre Datenbankverbindung und das Parsen von Daten aus Ihrer Datenbanktabelle in Python-Objekte.

• SQLAlchemy2: der größte Konkurrent in der Python-ORM-Welt. Persönlich gefällt mir die Syntax überhaupt nicht, aber jede für sich.

• TortoiseORM: Ich persönlich mag dieses von Django inspirierte asynchrone ORM wirklich; Es ist sauber und schön zu verwenden.

• Alternative ORMs: Es gibt viele wie Peewee, PonyORM usw.

Der Mittelweg: Abfrageersteller

Keine Datenbankverbindung. Geben Sie einfach Roh-SQL aus einer Python-basierten Abfrage aus und übergeben Sie es an den Datenbanktreiber.

• SQLAlchemy Core: der Kern-SQL-Abfrage-Builder, ohne den Teil „Zuordnung zu Objekten“. Es gibt auch ein darauf basierendes ORM höherer Ebene namens Datenbanken, das sehr gut aussieht. Ich frage mich jedoch, wie aktiv das Projekt entwickelt wird.

• PyPika: Ich weiß nicht viel darüber.

Einfaches Design: Datenbanktreiber

• asyncpg: Dies war der Goldstandard-Asynchron-Datenbanktreiber für Postgres, einer der ersten auf dem Markt und mit der höchsten Leistung. Während alle anderen Treiber die C-Bibliothek libpq als Schnittstelle zu Postgres verwenden, hat sich MagicStack dafür entschieden, ihre eigene benutzerdefinierte Implementierung neu zu schreiben und auch von der Python-DBAPI-Spezifikation abzuweichen. Wenn Leistung hier Ihr Hauptkriterium ist, dann ist asyncpg wahrscheinlich die beste Option.

• psycopg3: Nun, psycopg2 war eindeutig der König der synchronen Datenbanktreiberwelt für Python/Postgres. psycopg3 (umbenannt in einfach psycopg) ist die nächste, vollständig asynchrone Iteration dieser Bibliothek. Diese Bibliothek hat in den letzten Jahren wirklich an Bedeutung gewonnen und ich möchte näher darauf eingehen. Sehen Sie sich diesen interessanten Blog des Autors über die Anfänge von psycopg3 an.

Beachten Sie, dass es hier eindeutig eine breitere, konzeptionellere Diskussion über ORMs vs. Abfrage-Builder vs. Raw-SQL gibt. Darauf werde ich hier nicht eingehen.

Duplizierte Modelle

Pydantic ist mit FastAPI gebündelt und eignet sich hervorragend zum Modellieren, Validieren und Serialisieren von API-Antworten.

Wenn wir uns entscheiden, ein ORM zum Abrufen von Daten aus unserer Datenbank zu verwenden, ist es dann nicht etwas ineffizient, zwei Sätze von Datenbankmodellen synchron zu halten? (eine für das ORM, eine andere für Pydantic)?

Wäre es nicht großartig, wenn wir einfach Pydantic zum Modellieren der Datenbank verwenden könnten?

Genau dieses Problem hat der Ersteller von FastAPI mit der Bibliothek SQLModel zu lösen versucht.

Obwohl dies durchaus eine großartige Lösung für das Problem sein könnte, habe ich ein paar Bedenken:

• Wird dieses Projekt unter dem Single-Maintainer-Syndrom wie FastAPI leiden?

• Es ist noch ein relativ junges Projekt und Konzept, dessen Dokumentation nicht fantastisch ist.

• Es ist untrennbar mit Pydantic und SQLAlchemy verbunden, was bedeutet, dass eine Migration extrem schwierig wäre.

• Für komplexere Abfragen kann es erforderlich sein, zu SQLAlchemy weiter unten zu wechseln.

Zurück zu den Grundlagen

So viele Möglichkeiten! Analyselähmung.

Wenn es Unsicherheit gibt, würde ich den folgenden Grundsatz anwenden: Keep it simple.

SQL wurde vor 50 Jahren erfunden und ist immer noch eine Schlüsselkompetenz, die jeder Entwickler erlernen muss. Die Syntax ist durchweg leicht zu verstehen und für die meisten Anwendungsfälle unkompliziert zu schreiben (für die eingefleischten ORM-Benutzer da draußen: Probieren Sie es aus, Sie werden überrascht sein).

Verdammt, wir können heutzutage sogar Open-Source-LLMs verwenden, um (meistens funktionierende) SQL-Abfragen zu generieren und Ihnen das Tippen zu ersparen.

Während ORMs und Abfrage-Builder kommen und gehen, sind Datenbanktreiber wahrscheinlich konsistenter. Die ursprüngliche psycopg2-Bibliothek wurde vor fast 20 Jahren geschrieben und wird weltweit immer noch aktiv in der Produktion eingesetzt.

Verwendung von Psycopg mit Pydantic-Modellen

Wie bereits erwähnt, ist Psycopg möglicherweise nicht so leistungsfähig wie Asyncpg (die realen Auswirkungen dieser theoretischen Leistung sind jedoch umstritten), Psycopg konzentriert sich jedoch auf Benutzerfreundlichkeit und eine vertraute API.

Das Killer-Feature für mich sind Row Factories.

Mit dieser Funktionalität können Sie zurückgegebene Datenbankdaten jedem Python-Objekt zuordnen, einschließlich Standard-Lib-Datenklassen, Modellen aus der großartigen attrs-Bibliothek und ja, Pydantic-Modellen!

Für mich ist dies der beste Kompromiss an Ansätzen: die ultimative Flexibilität von Roh-SQL mit den Validierungs-/Typsicherheitsfunktionen von Pydantic zur Modellierung der Datenbank. Psycopg kümmert sich auch um Dinge wie die Bereinigung von Variableneingaben, um SQL-Injection zu vermeiden.

Es ist zu beachten, dass asyncpg auch die Zuordnung zu Pydantic-Modellen verarbeiten kann, allerdings eher als Workaround denn als integrierte Funktion. Weitere Informationen finden Sie in diesem Problemthread. Ich weiß auch nicht, ob dieser Ansatz gut mit anderen Modellierungsbibliotheken funktioniert.

Wie ich oben erwähnt habe, arbeite ich normalerweise mit Geodaten: ein Bereich, der von ORMs und Abfrageerstellern oft vernachlässigt wird. Durch den Wechsel zum Roh-SQL habe ich die Möglichkeit, Geodaten zu analysieren und zu entparsen, da ich akzeptablere Typen in reinem Python benötige. Siehe meinen entsprechenden Artikel zu diesem Thema.

Teil 2: Beispielverwendung

Erstellen Sie eine Datenbanktabelle

Hier erstellen wir eine einfache Datenbanktabelle namens „user“ in Roh-SQL.

Ich würde auch darüber nachdenken, die Datenbankerstellung und -migration nur mit SQL abzuwickeln, aber das ist ein Thema für einen anderen Artikel.

init_db.sql

CREATE TYPE public.userrole AS ENUM (
    'READ_ONLY',
    'STANDARD',
    'ADMIN'
);

CREATE TABLE public.users (
    id integer NOT NULL,
    username character varying,
    role public.userrole NOT NULL DEFAULT 'STANDARD',
    profile_img character varying,
    email_address character varying,
    is_email_verified boolean DEFAULT false,
    registered_at timestamp with time zone DEFAULT now()
);

Modellieren Sie Ihre Datenbank mit Pydantic

Hier erstellen wir ein Modell namens DbUser:

db_models.py

from typing import Optional
from enum import Enum
from datetime import datetime
from pydantic import BaseModel
from pydantic.functional_validators import field_validator
from geojson_pydantic import Feature

class UserRole(str, Enum):
    """Types of user, mapped to database enum userrole."""

    READ_ONLY = "READ_ONLY"
    STANDARD = "STANDARD"
    ADMIN = "ADMIN"

class DbUser(BaseModel):
    """Table users."""

    id: int
    username: str
    role: Optional[UserRole] = UserRole.STANDARD
    profile_img: Optional[str] = None
    email_address: Optional[str] = None
    is_email_verified: bool = False
    registered_at: Optional[datetime]
    # This is a geospatial type I will handle in the SQL
    favourite_place: Optional[dict]

    # DB computed fields (handled in the SQL)
    total_users: Optional[int] = None

    # This example isn't very realistic, but you get the idea
    @field_validator("is_email_verified", mode="before")
    @classmethod
    def i_want_my_ints_as_bools(cls, value: int) -> bool:
        """Example of a validator to convert data type."""
        return bool(value)

Hier erhalten wir die Typensicherheit und Validierung von Pydantic.

Wir können diesem Modell jede Form der Validierung oder Datentransformation hinzufügen, wenn die Daten aus der Datenbank extrahiert werden.

Psycopg mit FastAPI einrichten

Wir verwenden psycopg_pool, um eine gepoolte Datenbankverbindung zu erstellen:

db.py

from fastapi import Request
from psycopg import Connection
from psycopg_pool import AsyncConnectionPool

# You should be using environment variables in a settings file here
from app.config import settings


def get_db_connection_pool() -> AsyncConnectionPool:
    """Get the connection pool for psycopg.

    NOTE the pool connection is opened in the FastAPI server startup (lifespan).

    Also note this is also a sync `def`, as it only returns a context manager.
    """
    return AsyncConnectionPool(
        conninfo=settings.DB_URL.unicode_string(), open=False
    )


async def db_conn(request: Request) -> Connection:
    """Get a connection from the psycopg pool.

    Info on connections vs cursors:
    https://www.psycopg.org/psycopg3/docs/advanced/async.html

    Here we are getting a connection from the pool, which will be returned
    after the session ends / endpoint finishes processing.

    In summary:
    - Connection is created on endpoint call.
    - Cursors are used to execute commands throughout endpoint.
      Note it is possible to create multiple cursors from the connection,
      but all will be executed in the same db 'transaction'.
    - Connection is closed on endpoint finish.
    """
    async with request.app.state.db_pool.connection() as conn:
        yield conn

Als nächstes öffnen wir den Verbindungspool im FastAPI-Lebensdauerereignis:

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI

from .db import get_db_connection_pool

@asynccontextmanager
async def lifespan(app: FastAPI):
    """FastAPI startup/shutdown event."""
    # For this demo I use print, but please use logging!
    print("Starting up FastAPI server.")

    # Create a pooled db connection and make available in app state
    # NOTE we can access 'request.app.state.db_pool' in endpoints
    app.state.db_pool = get_db_connection_pool()
    await app.state.db_pool.open()

    yield

    # Shutdown events
    print("Shutting down FastAPI server.")
    # Here we make sure to close the connection pool
    await app.state.db_pool.close()

Wenn Sie nun die FastAPI-App starten, sollten Sie über einen offenen Verbindungspool verfügen, der bereit ist, Verbindungen von innerhalb von Endpunkten aufzunehmen.

Hilfsmethoden für das Pydantic-Modell

Es wäre nützlich, dem Pydantic-Modell einige Methoden für allgemeine Funktionen hinzuzufügen: einen Benutzer, alle Benutzer abrufen, einen Benutzer erstellen, einen Benutzer aktualisieren, einen Benutzer löschen.

Aber zuerst sollten wir einige Pydantic-Modelle für die Eingabevalidierung (um einen neuen Benutzer zu erstellen) und die Ausgabeserialisierung (Ihre JSON-Antwort über die API) erstellen.

user_schemas.py

from typing import Annotated
from pydantic import BaseModel, Field
from pydantic.functional_validators import field_validator
from geojson_pydantic import FeatureCollection, Feature, MultiPolygon, Polygon
from .db_models import DbUser

class UserIn(DbUser):
    """User details for insert into DB."""

    # Exclude fields not required for input
    id: Annotated[int, Field(exclude=True)] = None
    favourite_place: Optional[Feature]

    @field_validator("favourite_place", mode="before")
    @classmethod
    def parse_input_geojson(
        cls,
        value: FeatureCollection | Feature | MultiPolygon | Polygon,
    ) -> Optional[Polygon]:
        """Parse any format geojson into a single Polygon."""
        if value is None:
            return None
        # NOTE I don't include this helper function for brevity
        featcol = normalise_to_single_geom_featcol(value)
        return featcol.get("features")[0].get("geometry")

class UserOut(DbUser):
    """User details for insert into DB."""

    # Ensure it's parsed as a Polygon geojson from db object
    favourite_place: Polygon

    # More logic to append computed values

Dann können wir unsere Hilfsmethoden definieren: one, all, create:

db_models.py

...previous imports
from typing import Self, Optional
from fastapi.exceptions import HTTPException
from psycopg import Connection
from psycopg.rows import class_row

from .user_schemas import UserIn

class DbUser(BaseModel):
    """Table users."""

    ...the fields

    @classmethod
    async def one(cls, db: Connection, user_id: int) -> Self:
        """Get a user by ID.

        NOTE how the favourite_place field is converted in the db to geojson.
        """
        async with db.cursor(row_factory=class_row(cls)) as cur:
            sql = """
                SELECT
                    u.*,
                    ST_AsGeoJSON(favourite_place)::jsonb AS favourite_place,
                    (SELECT COUNT(*) FROM users) AS total_users
                FROM users u
                WHERE
                    u.id = %(user_id)s
                GROUP BY u.id;
            """

            await cur.execute(
                sql,
                {"user_id": user_id},
            )

            db_project = await cur.fetchone()
            if not db_project:
                raise KeyError(f"User ({user_identifier}) not found.")

            return db_project

    @classmethod
    async def all(
        cls, db: Connection, skip: int = 0, limit: int = 100
    ) -> Optional[list[Self]]:
        """Fetch all users."""
        async with db.cursor(row_factory=class_row(cls)) as cur:
            await cur.execute(
                """
                SELECT
                    *,
                    ST_AsGeoJSON(favourite_place)::jsonb
                FROM users
                OFFSET %(offset)s
                LIMIT %(limit)s;
                """,
                {"offset": skip, "limit": limit},
            )
            return await cur.fetchall()

    @classmethod
    async def create(
        cls,
        db: Connection,
        user_in: UserIn,
    ) -> Optional[Self]:
        """Create a new user."""

        # Omit defaults and empty values from the model
        model_dump = user_in.model_dump(exclude_none=True, exclude_default=True)
        columns = ", ".join(model_dump.keys())
        value_placeholders = ", ".join(f"%({key})s" for key in model_dump.keys())

        sql = f"""
            INSERT INTO users
                ({columns})
            VALUES
                ({value_placeholders})
            RETURNING *;
        """


        async with db.cursor(row_factory=class_row(cls)) as cur:
            await cur.execute(sql, model_dump)
            new_user = await cur.fetchone()

            if new_user is None:
                msg = f"Unknown SQL error for data: {model_dump}"
                print(f"Failed user creation: {model_dump}")
                raise HTTPException(status_code=500, detail=msg)

        return new_user

Verwendung

routes.py

from typing import Annotated
from fastapi import Depends, HTTPException
from psycopg import Connection

from .main import app
from .db import db_conn
from .models import DbUser
from .user_schemas import UserIn, UserOut

@app.post("/", response_model=UserOut)
async def create_user(
    user_info: UserIn,
    db: Annotated[Connection, Depends(db_conn)],
):
    """Create a new user.

    Here the input is parsed and validated by UserIn
    then the output is parsed and validated by UserOut
    returning the user json data.
    """

    new_user = await DbUser.create(db, user_info)
    if not new_user:
        raise HTTPException(
            status_code=422,
            detail="User creation failed.",
        )

    return new_user

    # NOTE within an endpoint we can also use
    # DbUser.one(db, user_id) and DbUser.all(db)

Diesen Ansatz habe ich begonnen, in einem von mir betreuten Projekt zu verwenden, dem FMTM, einem Tool zum Sammeln von Felddaten für Gemeinden auf der ganzen Welt.

Die vollständige Codebasis finden Sie hier.
Und ⭐ wenn Sie das nützlich fanden!

Das ist alles für den Moment! Ich hoffe, das hilft jemandem da draußen?

Das obige ist der detaillierte Inhalt vonFastAPI, Pydantic, Psycopgdie heilige Dreifaltigkeit für Python-Web-APIs. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!