FastAPI

FastAPI est un framework pour construire des API en Python.

Le générateur FastAPI crée une nouvelle application FastAPI avec une infrastructure AWS CDK configurée. 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

Générer une application FastAPI

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

VSCode

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
6. Cliquez sur Generate

CLI

pnpm

pnpm nx g @aws/nx-plugin:py#fast-api

yarn

yarn nx g @aws/nx-plugin:py#fast-api

npm

npx nx g @aws/nx-plugin:py#fast-api

bun

bunx nx g @aws/nx-plugin:py#fast-api

Vous pouvez également effectuer une simulation pour voir quels fichiers seraient modifiés

pnpm

pnpm nx g @aws/nx-plugin:py#fast-api --dry-run

yarn

yarn nx g @aws/nx-plugin:py#fast-api --dry-run

npm

npx nx g @aws/nx-plugin:py#fast-api --dry-run

bun

bunx nx g @aws/nx-plugin:py#fast-api --dry-run

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.
moduleName	string	-	Python module name

Note

Vous pouvez choisir entre une REST API ou une HTTP API pour déployer votre API. Veuillez consulter le API Gateway Developer Guide pour vous aider à décider.

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’application FastAPI

Le générateur créera également des constructs CDK pour déployer votre API, situés dans le répertoire packages/common/constructs.

Implémentation de votre 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. 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

Journalisation

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

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 :

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

Tracing

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

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

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 succès/échecs
• Métriques de cold start
• Métriques par route

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

Astuce

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

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

Modifications d’infrastructure

Comme 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 Function URL AWS Lambda. Pour cela, vous pouvez remplacer le construct généré common/constructs/src/app/apis/<name>-api.ts par un qui déploie une Function URL.

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',
        },
      },
    });
  }
}

Note

Pour un exemple complet, référez-vous au Tutoriel Dungeon Adventure

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 vendor OpenAPI x-streaming: true si vous comptez utiliser API Connection.

Par exemple, pour streamer une série d’objets JSON :

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

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 FastAPI

Le générateur FastAPI crée un construct CDK pour déployer votre API 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) {
    // Ajoute 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’API
2. Une API Gateway HTTP/REST comme déclencheur
3. Les rôles IAM et permissions
4. Un groupe de logs CloudWatch
5. La configuration de tracing X-Ray
6. Un namespace de métriques CloudWatch

Note

Si vous avez choisi l’authentification Cognito, vous devrez fournir la propriété identity au construct API :

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

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    const identity = new UserIdentity(this, 'Identity');

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

Le construct UserIdentity peut être généré avec le générateur ts#cloudscape-website-auth

Intégrations Type-Safe

Les constructeurs CDK d’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.

Intégrations par défaut

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

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

Accès aux intégrations

Vous pouvez accéder aux fonctions AWS Lambda sous-jacentes via la propriété integrations du construct d’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é en fonction des 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

Si vous souhaitez personnaliser les options utilisées lors de la création de la fonction 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

Vous pouvez également 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 d’intégration CDK approprié pour l’API HTTP ou REST. La méthode withOverrides est également typée. Par exemple, pour 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 lors de son accès 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
// de manière typée
api.integrations.getFile.bucket.grantRead(...);

Surcharge des autorisations

Vous pouvez également fournir des options dans votre intégration pour surcharger des options de méthode spécifiques comme les autorisations. Par exemple, pour utiliser l’authentification Cognito pour l’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

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 chaque opération nécessite un type d’intégration différent, ou si vous souhaitez 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

Si vous préférez déployer une seule fonction Lambda pour traiter toutes les requêtes de l’API, vous pouvez modifier librement 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 gestionnaire Lambda de routeur 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

Les opérations FastAPI étant définies en Python et l’infrastructure en TypeScript, nous utilisons une génération de code pour fournir des métadonnées au construct CDK afin d’offrir une interface type-safe.

Une cible generate:<ApiName>-metadata est ajoutée au project.json des constructs 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 versioning.

Note

Vous devrez exécuter un build après chaque modification de votre API pour maintenir les types à jour.

pnpm

pnpm nx run-many --target build --all

yarn

yarn nx run-many --target build --all

npm

npx nx run-many --target build --all

bun

bunx nx run-many --target build --all

Astuce

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 :

pnpm

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

yarn

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

npm

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

bun

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

Octroi d’accès (IAM uniquement)

Si vous avez choisi l’authentification IAM, vous pouvez utiliser grantInvokeAccess pour octroyer l’accès :

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

Développement local

Le générateur configure un serveur de développement local que vous pouvez lancer avec :

pnpm

pnpm nx run my-api:serve

yarn

yarn nx run my-api:serve

npm

npx nx run my-api:serve

bun

bunx nx run my-api:serve

Ceci démarre un serveur FastAPI local avec :

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

Invocation de votre FastAPI

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