FastAPI, Pydantic, Psycopgla sainte trinité pour les API Web Python

Partie 1 : Discussion

Entrez FastAPI

Tout d’abord, prenez le titre avec des pincettes.

Si je repartais de zéro avec le développement d'API web Python aujourd'hui, je regarderais probablement de plus près LiteStar, qui me semble être une meilleure architecture et avec une meilleure structure de gouvernance de projet.

Mais nous avons FastAPI et cela ne va pas bientôt aller nulle part. Je l'utilise pour de nombreux projets personnels et professionnels et j'apprécie toujours sa simplicité.

Pour un guide sur les modèles de conception FastAPI, ne cherchez pas plus loin que cette page.

Récupération des données de la base de données

Bien que FastAPI soit excellent dans la partie « API », il reste une incertitude persistante pour moi : comment accéder au mieux à la base de données, en particulier si nous devons également gérer des types de données géospatiales.

Revoyons nos options.

Remarque 1 : nous ne nous intéressons ici qu'aux bibliothèques async, car FastAPI est ASGI.

Remarque 2 : je ne discuterai que de la connexion à PostgreSQL, bien que certaines parties de la discussion soient toujours pertinentes pour d'autres bases de données.

Simple à coder | Conception complexe : ORM

Gère la connexion à votre base de données et l'analyse des données de votre table de base de données en objets Python.

• SQLAlchemy2 : le plus grand concurrent dans le monde Python ORM. Personnellement, je n'aime vraiment pas la syntaxe, mais chacun à sa manière.

• TortoiseORM : personnellement, j'aime beaucoup cet ORM asynchrone inspiré de Django ; c'est propre et agréable à utiliser.

• ORM alternatifs : il en existe beaucoup comme peewee, PonyORM, etc.

Le juste milieu : les générateurs de requêtes

Aucune connexion à la base de données. Générez simplement du SQL brut à partir d'une requête basée sur Python et transmettez-le au pilote de base de données.

• SQLAlchemy Core : le générateur de requêtes SQL de base, sans la partie mappage aux objets. Il existe également un ORM de niveau supérieur construit sur ces bases de données appelées qui a l'air très joli. Je me demande cependant dans quelle mesure le projet est activement développé.

• PyPika : Je ne sais pas grand chose sur celui-ci.

Conception simple : pilotes de base de données

• asyncpg : il s'agissait du pilote de base de données asynchrone de référence pour Postgres, étant l'un des premiers sur le marché et le plus performant. Alors que tous les autres pilotes utilisent la bibliothèque C libpq pour s'interfacer avec Postgres, MagicStack a choisi de réécrire sa propre implémentation personnalisée et de s'écarter également des spécifications Python DBAPI. Si la performance est votre principal critère ici, alors asyncpg est probablement la meilleure option.

• psycopg3 : eh bien, psycopg2 était clairement le roi du monde des pilotes de base de données synchrones pour Python/Postgres. psycopg3 (rebaptisé simplement psycopg) est la prochaine itération, entièrement asynchrone, de cette bibliothèque. Cette bibliothèque a vraiment pris tout son sens ces dernières années et je souhaite en discuter davantage. Voir ce blog intéressant de l'auteur sur les débuts de psycopg3.

Notez qu'il y a clairement une discussion plus large et plus conceptuelle à avoir ici autour des ORM, des générateurs de requêtes et du SQL brut. Je n'en parlerai pas ici.

Modèles dupliqués

Pydantic est fourni avec FastAPI et est excellent pour modéliser, valider et sérialiser les réponses API.

Si nous décidons d'utiliser un ORM pour récupérer des données de notre base de données, n'est-il pas un peu inefficace de synchroniser deux ensembles de modèles de base de données ? (un pour l'ORM, un autre pour Pydantic) ?

Ne serait-ce pas génial si nous pouvions simplement utiliser Pydantic pour modéliser la base de données ?

C'est exactement le problème que le créateur de FastAPI a essayé de résoudre avec la bibliothèque SQLModel.

Bien que cela puisse très bien être une excellente solution au problème, j'ai quelques inquiétudes :

• Ce projet souffrira-t-il du syndrome du responsable unique comme FastAPI ?

• C'est encore un projet et un concept relativement jeunes, où la documentation n'est pas fantastique.

• Il est intrinsèquement lié à Pydantic et SQLAlchemy, ce qui signifie que la migration serait extrêmement difficile.

• Pour les requêtes plus complexes, il peut être nécessaire de passer à SQLAlchemy ci-dessous.

Retour aux sources

Tant d'options ! Paralysie de l'analyse.

En cas d'incertitude, j'utiliserais le précepte suivant : restez simple.

SQL a été inventé il y a 50 ans et reste une compétence clé à acquérir pour tout développeur. Sa syntaxe est toujours facile à comprendre et simple à écrire pour la plupart des cas d'utilisation (pour les utilisateurs inconditionnels d'ORM, essayez-la, vous pourriez être surpris).

Bon sang, nous pouvons même utiliser des LLM open source de nos jours pour générer des requêtes SQL (pour la plupart fonctionnelles) et vous épargner la saisie.

Bien que les ORM et les générateurs de requêtes puissent aller et venir, les pilotes de base de données sont probablement plus cohérents. La bibliothèque psycopg2 originale a été écrite il y a près de 20 ans maintenant et est toujours activement utilisée en production à l'échelle mondiale.

Utiliser Psycopg avec des modèles Pydantic

Comme indiqué, même si psycopg n'est peut-être pas aussi performant qu'asyncpg (les implications réelles de cette performance théorique sont cependant discutables), psycopg se concentre sur la facilité d'utilisation et une API familière.

La fonctionnalité qui tue pour moi est Row Factories.

Cette fonctionnalité vous permet de mapper les données de base de données renvoyées à n'importe quel objet Python, y compris les classes de données lib standard, les modèles de la grande bibliothèque attrs et, oui, les modèles Pydantic !

Pour moi, c'est le meilleur compromis d'approches : la flexibilité ultime du SQL brut, avec les capacités de validation/sécurité de type de Pydantic pour modéliser la base de données. Psycopg gère également des choses comme l'assainissement des entrées variables pour éviter l'injection SQL.

Il convient de noter qu'asyncpg peut également gérer le mappage vers les modèles Pydantic, mais comme une solution de contournement plutôt qu'une fonctionnalité intégrée. Consultez ce fil de discussion pour plus de détails. Je ne sais pas non plus si cette approche fonctionne bien avec d'autres bibliothèques de modélisation.

Comme je l'ai mentionné ci-dessus, je travaille généralement avec des données géospatiales : un domaine souvent négligé par les ORM et les générateurs de requêtes. Passer au SQL brut me donne la possibilité d'analyser et d'analyser des données géospatiales car j'ai besoin de types plus acceptables en Python pur. Voir mon article connexe sur ce sujet.

Partie 2 : exemple d'utilisation

Créer une table de base de données

Ici, nous créons une simple table de base de données appelée utilisateur en SQL brut.

J'envisagerais également de gérer la création et les migrations de bases de données en utilisant uniquement SQL, mais c'est un sujet pour un autre article.

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()
);

Modélisez votre base de données avec Pydantic

Ici, nous créons un modèle appelé 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)

Ici, nous obtenons la sécurité de type et la validation de Pydantic.

Nous pouvons ajouter toute forme de validation ou de transformation de données à ce modèle lorsque les données sont extraites de la base de données.

Configuration de Psycopg avec FastAPI

Nous utilisons psycopg_pool pour créer une connexion à la base de données poolée :

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

Ensuite, nous ouvrons le pool de connexions dans l'événement de durée de vie FastAPI :

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()

Maintenant, lorsque votre application FastAPI démarre, vous devriez disposer d'un pool de connexions ouvert, prêt à établir une connexion depuis les points de terminaison internes.

Méthodes d'assistance pour le modèle Pydantic

Il serait utile d'ajouter quelques méthodes au modèle Pydantic pour des fonctionnalités communes : obtenir un utilisateur, tous les utilisateurs, créer un utilisateur, mettre à jour un utilisateur, supprimer un utilisateur.

Mais nous devons d'abord créer des modèles Pydantic pour la validation des entrées (pour créer un nouvel utilisateur) et la sérialisation des sorties (votre réponse JSON via l'API).

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

Ensuite, nous pouvons définir nos méthodes d'assistance : un, tous, créer :

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

Usage

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)

C'est l'approche que j'ai commencé à utiliser dans un projet que je gère, le FMTM, un outil permettant de collecter des données de terrain pour les communautés du monde entier.

Voir la base de code complète ici.
Et ⭐ si vous avez trouvé cela utile !

C'est tout pour le moment ! J'espère que cela aidera quelqu'un ?

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!