Aller au contenu
@aws/nx-plugin
Blog

FastAPI

FastAPI est un framework pour construire des APIs en Python.

Le générateur FastAPI crée une nouvelle application FastAPI avec une infrastructure AWS CDK ou Terraform. Le backend généré utilise AWS Lambda pour un déploiement serverless, exposé via une API AWS API Gateway. Il configure AWS Lambda Powertools pour l’observabilité, incluant la journalisation, le tracing AWS X-Ray et les métriques CloudWatch.

Utilisation

Section intitulée « Utilisation »

Générer une API FastAPI

Section intitulée « Générer une API FastAPI »

Vous pouvez générer une nouvelle API FastAPI de deux manières :

  1. Installez le Nx Console VSCode Plugin si ce n'est pas déjà fait
  2. Ouvrez la console Nx dans VSCode
  3. Cliquez sur Generate (UI) dans la section "Common Nx Commands"
  4. Recherchez @aws/nx-plugin - py#fast-api
  5. Remplissez les paramètres requis
    • Cliquez sur Generate

    Options

    Section intitulée « Options »
    Paramètre Type Par défaut Description
    name Requis string - Name of the API project to generate
    computeType string ServerlessApiGatewayRestApi The type of compute to use to deploy this API. Choose between ServerlessApiGatewayRestApi (default) or ServerlessApiGatewayHttpApi.
    auth string IAM The method used to authenticate with your API. Choose between IAM (default), Cognito or None.
    directory string packages The directory to store the application in.
    iacProvider string CDK The preferred IaC provider
    moduleName string - Python module name

    Sortie du générateur

    Section intitulée « Sortie du générateur »

    Le générateur créera la structure de projet suivante dans le répertoire <directory>/<api-name> :

    • project.json Configuration du projet et cibles de build
    • pyproject.toml Configuration du projet Python et dépendances
    • Répertoire<module_name>
      • __init__.py Initialisation du module
      • init.py Configure l’application FastAPI et le middleware powertools
      • main.py Implémentation de l’API
    • Répertoirescripts
      • generate_open_api.py Script pour générer un schéma OpenAPI depuis l’app FastAPI

    Infrastructure

    Section intitulée « Infrastructure »

    Ce générateur fournit de l’infrastructure as code basée sur votre iacProvider choisi. Il créera un projet dans packages/common qui inclut les constructions CDK ou modules Terraform pertinents.

    Le projet commun d’infrastructure as code est structuré comme suit :

    • Répertoirepackages/common/constructs
      • Répertoiresrc
        • Répertoireapp/ Constructions pour l’infrastructure spécifique à un projet/générateur
          • …
        • Répertoirecore/ Constructions génériques réutilisées par celles dans app
          • …
        • index.ts Point d’entrée exportant les constructions depuis app
      • project.json Cibles de build et configuration du projet

    Pour déployer votre API, les fichiers suivants sont générés :

    • Répertoirepackages/common/constructs/src
      • Répertoireapp
        • Répertoireapis
          • <project-name>.ts Construct CDK pour déployer votre API
      • Répertoirecore
        • Répertoireapi
          • http-api.ts Construct CDK pour déployer une API HTTP (si vous avez choisi de déployer une API HTTP)
          • rest-api.ts Construct CDK pour déployer une API REST (si vous avez choisi de déployer une API REST)
          • utils.ts Utilitaires pour les constructs d’API

    Implémentation de votre API FastAPI

    Section intitulée « Implémentation de votre API FastAPI »

    L’implémentation principale de l’API se trouve dans main.py. C’est ici que vous définissez vos routes API et leurs implémentations. Voici un exemple :

    from .init import app, tracer
    from pydantic import BaseModel
    

    class Item(BaseModel):
      name: str
    

    @app.get("/items/{item_id}")
    def get_item(item_id: int) -> Item:
        return Item(name=...)
    

    @app.post("/items")
    def create_item(item: Item):
        return ...

    Le générateur configure automatiquement plusieurs fonctionnalités :

    1. Intégration d’AWS Lambda Powertools pour l’observabilité
    2. Middleware de gestion d’erreurs
    3. Corrélation requête/réponse
    4. Collecte de métriques
    5. Handler AWS Lambda utilisant Mangum

    Observabilité avec AWS Lambda Powertools

    Section intitulée « Observabilité avec AWS Lambda Powertools »

    Journalisation

    Section intitulée « Journalisation »

    Le générateur configure la journalisation structurée avec AWS Lambda Powertools. Vous pouvez accéder au logger dans vos gestionnaires de route :

    from .init import app, logger
    

    @app.get("/items/{item_id}")
    def read_item(item_id: int):
        logger.info("Fetching item", extra={"item_id": item_id})
        return {"item_id": item_id}

    Le logger inclut automatiquement :

    • Des IDs de corrélation pour le tracing des requêtes
    • Le chemin et la méthode de la requête
    • Les informations de contexte Lambda
    • Les indicateurs de cold start

    Tracing

    Section intitulée « Tracing »

    Le tracing AWS X-Ray est configuré automatiquement. Vous pouvez ajouter des sous-segments personnalisés à vos traces :

    from .init import app, tracer
    

    @app.get("/items/{item_id}")
    @tracer.capture_method
    def read_item(item_id: int):
        # Crée un nouveau sous-segment
        with tracer.provider.in_subsegment("fetch-item-details"):
            # Votre logique ici
            return {"item_id": item_id}

    Métriques

    Section intitulée « Métriques »

    Les métriques CloudWatch sont collectées automatiquement pour chaque requête. Vous pouvez ajouter des métriques personnalisées :

    from .init import app, metrics
    from aws_lambda_powertools.metrics import MetricUnit
    

    @app.get("/items/{item_id}")
    def read_item(item_id: int):
        metrics.add_metric(name="ItemViewed", unit=MetricUnit.Count, value=1)
        return {"item_id": item_id}

    Les métriques par défaut incluent :

    • Le nombre de requêtes
    • Les compteurs de succès/échec
    • Les métriques de cold start
    • Les métriques par route

    Gestion des erreurs

    Section intitulée « Gestion des erreurs »

    Le générateur inclut une gestion d’erreurs complète :

    from fastapi import HTTPException
    

    @app.get("/items/{item_id}")
    def read_item(item_id: int):
        if item_id < 0:
            raise HTTPException(status_code=400, detail="Item ID must be positive")
        return {"item_id": item_id}

    Les exceptions non gérées sont capturées par le middleware et :

    1. Journalisent l’exception complète avec la stack trace
    2. Enregistrent une métrique d’échec
    3. Renvoient une réponse 500 sécurisée au client
    4. Préservent l’ID de corrélation

    Il est recommandé de spécifier des modèles de réponse pour vos opérations API afin d’améliorer la génération de code si vous utilisez le générateur api-connection. Voir ici pour plus de détails.

    Streaming

    Section intitulée « Streaming »

    Avec FastAPI, vous pouvez streamer une réponse au client avec le type de réponse StreamingResponse.

    Modifications d’infrastructure

    Section intitulée « Modifications d’infrastructure »

    Puisqu’AWS API Gateway ne supporte pas les réponses streamées, vous devrez déployer votre FastAPI sur une plateforme qui le supporte. L’option la plus simple est d’utiliser une URL de fonction Lambda AWS.

    Pour cela, vous pouvez remplacer le construct généré common/constructs/src/app/apis/<name>-api.ts par un qui déploie une URL de fonction.

    Exemple de construct FunctionURL pour le streaming
    import { Duration, Stack, CfnOutput } from 'aws-cdk-lib';
    import { IGrantable, Grant } from 'aws-cdk-lib/aws-iam';
    import {
      Runtime,
      Code,
      Tracing,
      LayerVersion,
      FunctionUrlAuthType,
      InvokeMode,
      Function,
    } from 'aws-cdk-lib/aws-lambda';
    import { Construct } from 'constructs';
    import url from 'url';
    import { RuntimeConfig } from '../../core/runtime-config.js';
    

    export class MyApi extends Construct {
      public readonly handler: Function;
    

      constructor(scope: Construct, id: string) {
        super(scope, id);
    

        this.handler = new Function(this, 'Handler', {
          runtime: Runtime.PYTHON_3_12,
          handler: 'run.sh',
          code: Code.fromAsset(
            url.fileURLToPath(
              new URL(
                '../../../../../../dist/packages/my_api/bundle',
                import.meta.url,
              ),
            ),
          ),
          timeout: Duration.seconds(30),
          tracing: Tracing.ACTIVE,
          environment: {
            AWS_CONNECTION_REUSE_ENABLED: '1',
          },
        });
    

        const stack = Stack.of(this);
        this.handler.addLayers(
          LayerVersion.fromLayerVersionArn(
            this,
            'LWALayer',
            `arn:aws:lambda:${stack.region}:753240598075:layer:LambdaAdapterLayerX86:24`,
          ),
        );
        this.handler.addEnvironment('PORT', '8000');
        this.handler.addEnvironment('AWS_LWA_INVOKE_MODE', 'response_stream');
        this.handler.addEnvironment('AWS_LAMBDA_EXEC_WRAPPER', '/opt/bootstrap');
        const functionUrl = this.handler.addFunctionUrl({
          authType: FunctionUrlAuthType.AWS_IAM,
          invokeMode: InvokeMode.RESPONSE_STREAM,
          cors: {
            allowedOrigins: ['*'],
            allowedHeaders: [
              'authorization',
              'content-type',
              'x-amz-content-sha256',
              'x-amz-date',
              'x-amz-security-token',
            ],
          },
        });
    

        new CfnOutput(this, 'MyApiUrl', { value: functionUrl.url });
    

        // Enregistre l'URL de l'API dans la configuration runtime pour la découverte client
        RuntimeConfig.ensure(this).config.apis = {
          ...RuntimeConfig.ensure(this).config.apis!,
          MyApi: functionUrl.url,
        };
      }
    

      public grantInvokeAccess(grantee: IGrantable) {
        Grant.addToPrincipal({
          grantee,
          actions: ['lambda:InvokeFunctionUrl'],
          resourceArns: [this.handler.functionArn],
          conditions: {
            StringEquals: {
              'lambda:FunctionUrlAuthType': 'AWS_IAM',
            },
          },
        });
      }
    }

    Implémentation

    Section intitulée « Implémentation »

    Une fois l’infrastructure mise à jour pour supporter le streaming, vous pouvez implémenter une API streaming dans FastAPI. L’API doit :

    • Retourner une StreamingResponse
    • Déclarer le type de retour de chaque chunk de réponse
    • Ajouter l’extension OpenAPI x-streaming: true si vous prévoyez d’utiliser API Connection.

    Par exemple, si vous souhaitez streamer une série d’objets JSON depuis votre API, vous pouvez l’implémenter ainsi :

    from pydantic import BaseModel
    from fastapi.responses import StreamingResponse
    

    class Chunk(BaseModel):
      message: str
      timestamp: datetime
    

    async def stream_chunks():
      for i in range(0, 100):
        yield Chunk(message=f"This is chunk {i}", timestamp=datetime.now())
    

    @app.get("/stream", openapi_extra={'x-streaming': True})
    def my_stream() -> Chunk:
        return StreamingResponse(stream_chunks(), media_type="application/json")

    Consommation

    Section intitulée « Consommation »

    Pour consommer un flux de réponses, vous pouvez utiliser le Générateur API Connection qui fournira une méthode typée pour itérer sur les chunks streamés.

    Déploiement de votre API FastAPI

    Section intitulée « Déploiement de votre API FastAPI »

    Le générateur FastAPI crée une infrastructure CDK ou Terraform en fonction du iacProvider sélectionné. Vous pouvez l’utiliser pour déployer votre API.

    Le construct CDK pour déployer votre API se trouve dans le dossier common/constructs. Vous pouvez l’utiliser dans une application CDK :

    import { MyApi } from ':my-scope/common-constructs';
    

    export class ExampleStack extends Stack {
      constructor(scope: Construct, id: string) {
        // Ajoutez l'API à votre stack
        const api = new MyApi(this, 'MyApi', {
          integrations: MyApi.defaultIntegrations(this).build(),
        });
      }
    }

    Ceci configure :

    1. Une fonction AWS Lambda pour chaque opération de l’application FastAPI
    2. Une API Gateway HTTP/REST comme déclencheur
    3. Les rôles et permissions IAM
    4. Un groupe de logs CloudWatch
    5. La configuration de tracing X-Ray
    6. Un espace de noms pour les métriques CloudWatch

    Intégrations

    Section intitulée « Intégrations »

    Les constructs CDK pour l’API REST/HTTP sont configurés pour fournir une interface typée permettant de définir des intégrations pour chacune de vos opérations.

    Les constructs CDK offrent un support complet d’intégration typée comme décrit ci-dessous.

    Intégrations par défaut

    Section intitulée « Intégrations par défaut »

    Vous pouvez utiliser la méthode statique defaultIntegrations pour utiliser le modèle par défaut, qui définit une fonction AWS Lambda individuelle pour chaque opération :

    new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this).build(),
    });

    Accès aux intégrations

    Section intitulée « Accès aux intégrations »

    Vous pouvez accéder aux fonctions AWS Lambda sous-jacentes via la propriété integrations du construct API, de manière typée. Par exemple, si votre API définit une opération nommée sayHello et que vous devez ajouter des permissions à cette fonction, vous pouvez le faire comme suit :

    const api = new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this).build(),
    });
    

    // sayHello est typé selon les opérations définies dans votre API
    api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({
      effect: Effect.ALLOW,
      actions: [...],
      resources: [...],
    }));

    Personnalisation des options par défaut

    Section intitulée « Personnalisation des options par défaut »

    Si vous souhaitez personnaliser les options utilisées lors de la création des fonctions Lambda pour chaque intégration par défaut, vous pouvez utiliser la méthode withDefaultOptions. Par exemple, si vous voulez que toutes vos fonctions Lambda résident dans un VPC :

    const vpc = new Vpc(this, 'Vpc', ...);
    

    new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this)
        .withDefaultOptions({
          vpc,
        })
        .build(),
    });

    Surcharge des intégrations

    Section intitulée « Surcharge des intégrations »

    Vous pouvez aussi surcharger les intégrations pour des opérations spécifiques en utilisant la méthode withOverrides. Chaque surcharge doit spécifier une propriété integration typée selon le construct CDK d’intégration approprié pour l’API HTTP ou REST. La méthode withOverrides est également typée. Par exemple, si vous voulez rediriger une API getDocumentation vers une documentation hébergée sur un site externe :

    new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this)
        .withOverrides({
          getDocumentation: {
            integration: new HttpIntegration('https://example.com/documentation'),
          },
        })
        .build(),
    });

    Vous remarquerez que l’intégration surchargée n’a plus de propriété handler lorsqu’on y accède via api.integrations.getDocumentation.

    Vous pouvez ajouter des propriétés supplémentaires à une intégration qui seront également typées, permettant d’abstraire d’autres types d’intégration tout en conservant le typage. Par exemple, si vous avez créé une intégration S3 pour une API REST et que vous souhaitez référencer le bucket pour une opération particulière :

    const storageBucket = new Bucket(this, 'Bucket', { ... });
    

    const apiGatewayRole = new Role(this, 'ApiGatewayS3Role', {
      assumedBy: new ServicePrincipal('apigateway.amazonaws.com'),
    });
    

    storageBucket.grantRead(apiGatewayRole);
    

    const api = new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this)
        .withOverrides({
          getFile: {
            bucket: storageBucket,
            integration: new AwsIntegration({
              service: 's3',
              integrationHttpMethod: 'GET',
              path: `${storageBucket.bucketName}/{fileName}`,
              options: {
                credentialsRole: apiGatewayRole,
                requestParameters: {
                  'integration.request.path.fileName': 'method.request.querystring.fileName',
                },
                integrationResponses: [{ statusCode: '200' }],
              },
            }),
            options: {
              requestParameters: {
                'method.request.querystring.fileName': true,
              },
              methodResponses: [{
                statusCode: '200',
              }],
            }
          },
        })
        .build(),
    });
    

    // Plus tard, peut-être dans un autre fichier, vous pouvez accéder à la propriété bucket
    // que nous avons définie de manière typée
    api.integrations.getFile.bucket.grantRead(...);

    Surcharge des autorisations

    Section intitulée « Surcharge des autorisations »

    Vous pouvez aussi fournir des options dans votre intégration pour surcharger des options de méthode spécifiques comme les autorisations. Par exemple, si vous souhaitez utiliser l’authentification Cognito pour votre opération getDocumentation :

    new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this)
        .withOverrides({
          getDocumentation: {
            integration: new HttpIntegration('https://example.com/documentation'),
            options: {
              authorizer: new CognitoUserPoolsAuthorizer(...) // pour REST, ou HttpUserPoolAuthorizer pour une API HTTP
            }
          },
        })
        .build(),
    });

    Intégrations explicites

    Section intitulée « Intégrations explicites »

    Si vous préférez, vous pouvez choisir de ne pas utiliser les intégrations par défaut et fournir directement une intégration pour chaque opération. Ceci est utile si, par exemple, chaque opération nécessite un type d’intégration différent ou si vous voulez obtenir une erreur de type lors de l’ajout de nouvelles opérations :

    new MyApi(this, 'MyApi', {
      integrations: {
        sayHello: {
          integration: new LambdaIntegration(...),
        },
        getDocumentation: {
          integration: new HttpIntegration(...),
        },
      },
    });

    Modèle Router

    Section intitulée « Modèle Router »

    Si vous préférez déployer une seule fonction Lambda pour gérer toutes les requêtes API, vous pouvez librement modifier la méthode defaultIntegrations de votre API pour créer une seule fonction au lieu d’une par intégration :

    packages/common/constructs/src/app/apis/my-api.ts
    export class MyApi<...> extends ... {
    

      public static defaultIntegrations = (scope: Construct) => {
        const router = new Function(scope, 'RouterHandler', { ... });
        return IntegrationBuilder.rest({
          ...
          defaultIntegrationOptions: {},
          buildDefaultIntegration: (op) => {
            return {
              # Référencer le même handler router dans chaque intégration
              integration: new LambdaIntegration(router),
            };
          },
        });
      };
    }

    Vous pouvez modifier le code selon vos préférences, par exemple définir la fonction router comme paramètre de defaultIntegrations au lieu de la construire dans la méthode.

    Génération de code

    Section intitulée « Génération de code »

    Comme les opérations FastAPI sont définies en Python et l’infrastructure CDK en TypeScript, nous instrumentons la génération de code pour fournir des métadonnées au construct CDK afin d’offrir une interface typée pour les intégrations.

    Une cible generate:<ApiName>-metadata est ajoutée au project.json des constructs communs pour faciliter cette génération, produisant un fichier comme packages/common/constructs/src/generated/my-api/metadata.gen.ts. Ce fichier étant généré au build, il est ignoré dans le contrôle de version.

    Si vous travaillez simultanément sur l’infrastructure CDK et FastAPI, vous pouvez utiliser nx watch pour régénérer ces types à chaque modification :

    Terminal window
    pnpm nx watch --projects=<FastAPIProject> -- \
    pnpm nx run <InfraProject>:"generate:<ApiName>-metadata"

    Octroi d’accès (IAM uniquement)

    Section intitulée « Octroi d’accès (IAM uniquement) »

    Si vous avez choisi l’authentification IAM, vous pouvez utiliser la méthode grantInvokeAccess pour octroyer l’accès à votre API :

    api.grantInvokeAccess(myIdentityPool.authenticatedRole);

    Développement local

    Section intitulée « Développement local »

    Le générateur configure un serveur de développement local exécutable avec :

    Terminal window
    pnpm nx run my-api:serve

    Ceci démarre un serveur de développement FastAPI avec :

    • Rechargement automatique sur modification
    • Documentation interactive de l’API sur /docs ou /redoc
    • Schéma OpenAPI sur /openapi.json

    Invocation de votre API FastAPI

    Section intitulée « Invocation de votre API FastAPI »

    Pour invoquer votre API depuis un site React, vous pouvez utiliser le générateur api-connection.