TI – Blog de Frank-Mich

Histoire courte, j’ai un API qui n’est pas sous FastAPI, mais je me sers quand même de FastAPI pour générer la documentation au format d’OpenAPI.

J’ai donc des modèles pour les points de terminaisons et je souhaite qu’ils fonctionnent avec et sans FastAPI, car dans leur version pure, FastAPI n’est pas déployé. (Pour faire une autre histoire courte, installer FastAPI alourdi un peu le paquet à déployer, mais surtout les démarrage à froid des lambdas AWS à cause du délai d’initialisation de Sentry. Sentry détecte que FastAPI est présent dans l’environnement et ajoute automatiquement du monitoring, ce qui prend un bon 500ms supplémentaire. Mesuré en décembre 2023.)

Bref, en m’inspirant fortement d’une réponse sur StackOverflow, j’ai produit ceci.

# models.py
from typing import Annotated, Any, cast

from pydantic import BaseModel, Field
from pydantic.fields import FieldInfo

try:
    from fastapi import Query  # type:ignore[import-not-found]

    def FieldQuery(*args: Any, **kwargs: Any) -> FieldInfo:  # noqa:N802
        return cast(FieldInfo, Field(Query(*args, **kwargs)))

except ImportError:

    def FieldQuery(*args: Any, **kwargs: Any) -> FieldInfo:  # noqa:N802
        return cast(FieldInfo, Field(*args, **kwargs))  # type: ignore[pydantic-field]


class QueryParams(BaseModel):
    limit: Annotated[
        int,
        FieldQuery(
            -1,
            description="The limit of values to fetch. -1 means no limit.",
        ),
    ]

Puis, pour la route

# routes.py
from fastapi import FastAPI

from models import QueryParams

app = FastAPI()

@app.get("/things")
def get_things(
    params: Annotated[QueryParams, Depends()],
) -> dict:
    ...

Est-ce parfait? Avec les «type: ignore», «noqa», «import try/except», pas dutout. Ça ressemble à un gros hack.

Est-ce que ça fonctionne? Oh que oui!

Note: Les tags «noqa» et «type: ignore» sont respectivement liés au linting avec ruff et mypy.

import os from http import HTTPStatus from typing import Annotated import stripe from fastapi import Depends, FastAPI, Header, HTTPException, Request app = FastAPI() async def get_body(request: Request) -> bytes: return await request.body() @app.post("/webhook", status_code=HTTPStatus.NO_CONTENT) def post_report( stripe_signature: Annotated[str, Header(alias="stripe-signature")], body: bytes = Depends(get_body), ) -> None: endpoint_secret = os.environ["ENDPOINT_SECRET"] try: # signature validation event = stripe.Webhook.construct_event(body, stripe_signature, endpoint_secret) except ValueError as e: # Invalid payload raise HTTPException(status_code=HTTPStatus.BAD_REQUEST) from e except stripe.error.SignatureVerificationError as e: # Invalid signature raise HTTPException(status_code=HTTPStatus.UNPROCESSABLE_ENTITY) from e print(event) return

Étiquette : TI

Modèle hybride Pydantic/FastAPI

Modèle de FastAPI Stripe Webhook