Aller au contenu

FastAPI

Filter this guide Pick generator option values to hide sections that don't apply.

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 traçage AWS X-Ray et les métriques CloudWatch.

Vous pouvez générer une nouvelle application 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#api
  5. Remplissez les paramètres requis
    • framework: fastapi
  6. Cliquez sur Generate
Paramètre Type Par défaut Description
name Requis string - Nom du projet API à générer
framework fastapi fastapi Le framework d'API à utiliser.
integrationPattern isolated | shared isolated Comment les intégrations API Gateway sont générées pour l'API. Choisissez entre isolated (par défaut) et shared.
auth iam | cognito | custom iam La méthode utilisée pour s'authentifier auprès de votre API. Choisissez entre iam (par défaut), cognito ou custom.
directory string packages Le répertoire dans lequel stocker l'application.
subDirectory string - Le sous-répertoire dans lequel le projet est placé. Par défaut, il s'agit du nom du projet.
iac inherit | cdk | terraform inherit Le fournisseur IaC préféré. Par défaut, celui-ci est hérité de votre sélection initiale.
moduleName string - Nom du module Python
infra rest-lambda | http-lambda | none rest-lambda Le type d'infrastructure à utiliser pour déployer cette API.
preferInstallDependencies boolean true Indique s'il faut privilégier l'installation des dépendances après l'exécution du générateur. Définir sur false pour différer l'installation lors de l'exécution de plusieurs générateurs en lot (une installation s'exécute quand même si nécessaire pour que les générateurs suivants puissent calculer le graphe de projet Nx) ; installer une seule fois à la fin.

Le générateur va créer 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
  • run.sh Script de bootstrap Lambda Web Adapter pour démarrer l’application FastAPI via uvicorn
  • 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

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

L’application déployée a l’architecture suivante :

ClientWAFAPI Gateway(REST API)LambdaCloudWatch(Logs, Metrics)X-Ray(Traces)

Les API REST incluent une Web ACL AWS WAFv2 devant l’étape API Gateway avec l’ensemble de règles par défaut géré par AWS activé.

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 pydantic import BaseModel
from .init import app, tracer
class Item(BaseModel):
name: str
@app.get("/items/{item_id}")
@tracer.capture_method
def get_item(item_id: int) -> Item:
return Item(name=...)
@app.post("/items")
@tracer.capture_method
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 des erreurs
  3. Corrélation des requêtes/réponses
  4. Collecte de métriques
  5. Déploiement AWS Lambda via Lambda Web Adapter avec uvicorn
  6. Streaming typé (API REST uniquement)

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 traçage des requêtes
  • Le chemin et la méthode de la requête
  • Les informations de contexte Lambda
  • Des indicateurs de démarrage à froid

Le traçage 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}

Des 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 démarrage à froid
  • Des métriques par route

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

Lorsque votre API est protégée par authentification, vos gestionnaires de route ont souvent besoin de savoir qui appelle. La FastAPI générée s’exécute dans AWS Lambda via le Lambda Web Adapter, qui transmet le contexte de requête API Gateway sous forme de JSON dans l’en-tête x-amzn-request-context. Vous pouvez le lire depuis la Request FastAPI pour extraire l’identité de l’appelant.

À titre d’exemple, ajoutons un endpoint /me qui retourne les détails de l’utilisateur appelant. Nous implémenterons l’extraction comme une dépendance FastAPI afin qu’elle puisse être réutilisée entre les routes. La forme du contexte de requête — et donc la façon dont vous extrayez l’identité — dépend à la fois de votre méthode auth sélectionnée et du fait que vous ayez déployé une API REST ou HTTP.

auth = iam

Pour l’authentification IAM, nous recherchons l’appelant dans Cognito en utilisant le sub extrait du contexte de requête API Gateway. Créez identity.py à côté de main.py :

import json
import os
from typing import Annotated
from boto3 import client
from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel
cognito = client("cognito-idp")
class Identity(BaseModel):
sub: str
username: str
def get_identity(request: Request) -> Identity:
# Le Lambda Web Adapter transmet le contexte de requête API Gateway sous forme de JSON
request_context_header = request.headers.get("x-amzn-request-context")
if not request_context_header:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
request_context = json.loads(request_context_header)
provider = request_context.get("identity", {}).get("cognitoAuthenticationProvider")
sub = provider.split(":")[-1] if provider else None
if not sub:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
users = cognito.list_users(
# Suppose que l'ID du user pool est configuré dans l'environnement lambda
UserPoolId=os.environ["USER_POOL_ID"],
Limit=1,
Filter=f'sub="{sub}"',
).get("Users", [])
if len(users) != 1:
raise HTTPException(status_code=403, detail=f"No user found with subjectId {sub}")
return Identity(sub=sub, username=users[0]["Username"])
CurrentUser = Annotated[Identity, Depends(get_identity)]
auth = cognito

Avec auth: 'cognito', l’autorisateur Cognito User Pools d’API Gateway vérifie le JWT que l’appelant fournit dans l’en-tête Authorization et place les claims vérifiés sur le contexte de requête.

Créez identity.py à côté de main.py :

import json
from typing import Annotated
from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel
class Identity(BaseModel):
sub: str
username: str
def get_identity(request: Request) -> Identity:
# Le Lambda Web Adapter transmet le contexte de requête API Gateway sous forme de JSON
request_context_header = request.headers.get("x-amzn-request-context")
if not request_context_header:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
request_context = json.loads(request_context_header)
claims = request_context.get("authorizer", {}).get("claims", {})
sub = claims.get("sub")
username = claims.get("username")
if not sub or not username:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
return Identity(sub=sub, username=username)
CurrentUser = Annotated[Identity, Depends(get_identity)]

Vous pouvez ensuite injecter la dépendance CurrentUser dans n’importe quelle route qui a besoin de l’identité de l’appelant :

from .identity import CurrentUser, Identity
from .init import app, tracer
@app.get("/me")
@tracer.capture_method
def me(identity: CurrentUser) -> Identity:
return identity
infra = rest-lambda

La FastAPI générée supporte les réponses streamées nativement lors de l’utilisation d’une API REST. L’infrastructure est configurée pour utiliser le AWS Lambda Web Adapter afin d’exécuter votre FastAPI via uvicorn dans Lambda, avec ResponseTransferMode.STREAM dans API Gateway pour toutes les opérations d’API REST, ce qui permet au streaming de fonctionner aux côtés des opérations non-streamées.

Le fichier init.py généré exporte une classe JsonStreamingResponse qui fournit un streaming typé avec une génération de schéma OpenAPI appropriée. Cela garantit que le générateur connection peut produire des méthodes client de streaming correctement typées.

from pydantic import BaseModel
from .init import app, JsonStreamingResponse
class Chunk(BaseModel):
message: str
async def generate_chunks():
for i in range(100):
yield Chunk(message=f"This is chunk {i}")
@app.post(
"/stream",
response_class=JsonStreamingResponse,
responses={200: JsonStreamingResponse.openapi_response(Chunk, "Stream of chunks")},
)
async def my_stream() -> JsonStreamingResponse:
return JsonStreamingResponse(generate_chunks())

La classe JsonStreamingResponse :

  1. Sérialise les modèles Pydantic au format JSON Lines (application/jsonl)
  2. Fournit un helper openapi_response qui génère le schéma OpenAPI correct avec itemSchema, permettant au générateur connection de produire des méthodes client de streaming typées

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

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 FastAPI.

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) {
// 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’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 traçage X-Ray
  6. Un espace de noms pour les métriques CloudWatch
auth = cognito
auth = custom
infra = rest-lambda

Pour les API REST, le construct généré associe par défaut une Web ACL AWS WAFv2 à l’étape API Gateway. La Web ACL utilise l’ensemble de règles par défaut géré par AWS (AWSManagedRulesCommonRuleSet et AWSManagedRulesKnownBadInputsRuleSet), offrant une protection contre les exploits web courants, y compris le Top 10 OWASP. Les journaux de requêtes WAF sont écrits dans un groupe CloudWatch Logs.

Vous pouvez modifier le construct rest-api généré pour ajouter, supprimer ou ajuster des règles (par exemple, pour ajouter des règles basées sur le taux ou des groupes de règles gérées supplémentaires).

Pour désactiver cette fonctionnalité (par exemple, pour attacher votre propre Web ACL), définissez enableWaf sur false :

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
enableWaf: false,
});
infra = rest-lambda

Pour les API REST, l’infrastructure générée active la journalisation des accès par défaut, écrivant une ligne JSON structurée par requête dans un groupe CloudWatch Logs dédié. Le groupe de journaux est chiffré avec une clé KMS gérée par le client et conservé pendant un an.

API Gateway écrit les journaux d’accès en utilisant un rôle CloudWatch Logs au niveau du compte. Ce rôle est configuré sur le paramètre AWS::ApiGateway::Account, qui est un singleton par région par compte — il n’y a qu’un seul rôle pour chaque API REST dans la région. Pour gérer cela en toute sécurité à travers plusieurs stacks déployées indépendamment, l’infrastructure générée :

  • Crée un rôle CloudWatch Logs partagé et le configure sur le compte uniquement lorsqu’aucun rôle fonctionnel n’est déjà défini, de sorte que les déploiements n’écrasent jamais un rôle appartenant à une autre stack.
  • Laisse le paramètre du compte intact lors du démontage, de sorte que la destruction d’une stack ne désactive jamais la journalisation pour les autres API REST dans la région.

Le rôle du compte est géré par le construct ApiGatewayAccount, un singleton à portée de stack résolu via ApiGatewayAccount.ensure(scope). Chaque stage d’API REST en dépend, et le rôle est configuré par une ressource personnalisée basée sur Lambda.

Vous pouvez personnaliser le format du journal d’accès en passant deployOptions lors de la construction de votre API :

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

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.

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

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: [...],
}));

Si votre API utilise le modèle shared, la fonction Lambda router partagée est exposée via api.integrations.$router :

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
});
api.integrations.$router.handler.addEnvironment('LOG_LEVEL', 'DEBUG');

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

Pour personnaliser les options utilisées pour créer l’intégration par défaut pour des opérations spécifiques (sans affecter les autres), vous pouvez utiliser la méthode withOperationOptions. Par exemple, si vous souhaitez augmenter le timeout de la fonction Lambda pour une seule opération :

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this)
.withOperationOptions({
sayHello: {
timeout: Duration.seconds(60),
},
})
.build(),
});
// Les opérations sélectionnées restent des intégrations par défaut, elles sont donc toujours typées en conséquence :
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({ ... }));

Les options que vous spécifiez sont fusionnées avec les options d’intégration par défaut (et toutes les options définies via withDefaultOptions). Notez que vous ne pouvez pas spécifier d’options pour les opérations que vous avez remplacées via withOverrides, car celles-ci n’utilisent plus l’intégration par défaut.

Vous rencontrerez une erreur de type si la même opération est ciblée à la fois par withOperationOptions et withOverrides, quel que soit l’ordre dans lequel vous les appelez.

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

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

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

Les constructs d’API CDK générés supportent deux modèles d’intégration :

  • isolated crée une fonction Lambda par opération. C’est le modèle par défaut pour les API générées.
  • shared crée une seule fonction Lambda router par défaut et la réutilise pour chaque opération sauf si vous surchargez des intégrations spécifiques.

isolated vous donne des permissions et une configuration plus granulaires par opération. shared réduit la prolifération des fonctions Lambda et des intégrations API Gateway tout en permettant des surcharges sélectives.

Par exemple, définir pattern sur 'shared' crée une fonction unique 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) => {
...
return IntegrationBuilder.rest({
pattern: 'shared',
...
});
};
}

Puisque 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.

auth = iam

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

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

Terminal window
pnpm nx serve my-api

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

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

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

Utilisez le générateur connection pour intégrer ce projet avec d’autres dans votre espace de travail. Les connexions suivantes impliquent ce projet :

FastAPI
React vers FastAPI Appeler une API Python FastAPI depuis un site React
FastAPI Amazon DynamoDB Python
FastAPI vers Python DynamoDB Connecter une FastAPI à une table DynamoDB