Ir al contenido

FastAPI

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

FastAPI es un framework para construir APIs en Python.

El generador de FastAPI crea una nueva FastAPI con configuración de infraestructura usando AWS CDK o Terraform. El backend generado utiliza AWS Lambda para despliegue serverless, expuesto a través de una API de AWS API Gateway. Configura AWS Lambda Powertools para observabilidad, incluyendo registro, trazado con AWS X-Ray y métricas de CloudWatch.

Puedes generar una nueva API FastAPI de dos maneras:

  1. Instale el Nx Console VSCode Plugin si aún no lo ha hecho
  2. Abra la consola Nx en VSCode
  3. Haga clic en Generate (UI) en la sección "Common Nx Commands"
  4. Busque @aws/nx-plugin - py#api
  5. Complete los parámetros requeridos
    • framework: fastapi
  6. Haga clic en Generate
ParámetroTipoPredeterminadoDescripción
name Requeridostring-Nombre del proyecto de API a generar
framework fastapifastapiEl framework de API a utilizar.
integrationPattern isolated | sharedisolatedCómo se generan las integraciones de API Gateway para la API. Elija entre isolated (predeterminado) y shared.
auth iam | cognito | customiamEl método utilizado para autenticar con tu API. Elige entre iam (predeterminado), cognito o custom.
directory stringpackagesEl directorio donde almacenar la aplicación.
subDirectory string-El subdirectorio en el que se coloca el proyecto. Por defecto, este es el nombre del proyecto.
iac inherit | cdk | terraforminheritEl proveedor de IaC preferido. Por defecto, se hereda de tu selección inicial.
moduleName string-Nombre del módulo Python
infra rest-lambda | http-lambda | nonerest-lambdaEl tipo de infraestructura a utilizar para desplegar esta API.
preferInstallDependencies booleantrueSi se prefiere instalar las dependencias después de que se ejecute el generador. Establecer en false para diferir la instalación al procesar múltiples generadores en lote (la instalación aún se ejecuta si es necesario para que los generadores subsecuentes puedan calcular el grafo de proyectos de Nx); instalar una vez al final.

El generador creará la siguiente estructura de proyecto en el directorio <directory>/<api-name>:

  • project.json Configuración del proyecto y objetivos de build
  • pyproject.toml Configuración del proyecto Python y dependencias
  • run.sh Script de arranque de Lambda Web Adapter para iniciar la app FastAPI mediante uvicorn
  • Directorio<module_name>
    • __init__.py Inicialización del módulo
    • init.py Configura la aplicación FastAPI y el middleware de powertools
    • main.py Implementación de la API
  • Directorioscripts
    • generate_open_api.py Script para generar un esquema OpenAPI desde la app FastAPI

Dado que este generador proporciona infraestructura como código basada en tu proveedor de iacProvider seleccionado, creará un proyecto en packages/common que incluye los constructos CDK o módulos de Terraform correspondientes.

El proyecto común de infraestructura como código tiene la siguiente estructura:

  • Directoriopackages/common/constructs
    • Directoriosrc
      • Directorioapp/ Constructos para infraestructura específica de un proyecto/generador
      • Directoriocore/ Constructos genéricos reutilizados por los constructos en app
      • index.ts Punto de entrada que exporta los constructos de app
    • project.json Objetivos de compilación y configuración del proyecto

Para implementar tu API, se generan los siguientes archivos:

  • Directoriopackages/common/constructs/src
    • Directorioapp
      • Directorioapis
        • <project-name>.ts Constructo de CDK para implementar tu API
    • Directoriocore
      • Directorioapi
        • http-api.ts Constructo de CDK para implementar una API HTTP (si seleccionaste implementar una API HTTP)
        • rest-api.ts Constructo de CDK para implementar una API REST (si seleccionaste implementar una API REST)
        • utils.ts Utilidades para los constructos de API

La aplicación desplegada tiene la siguiente arquitectura:

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

Las REST APIs incluyen una Web ACL de AWS WAFv2 frente al stage de API Gateway con el conjunto de reglas predeterminado administrado por AWS habilitado.

La implementación principal de la API está en main.py. Aquí es donde defines las rutas de tu API y sus implementaciones. Aquí un ejemplo:

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

El generador configura automáticamente varias características:

  1. Integración de AWS Lambda Powertools para observabilidad
  2. Middleware de manejo de errores
  3. Correlación de solicitudes/respuestas
  4. Recolección de métricas
  5. Despliegue en AWS Lambda mediante Lambda Web Adapter con uvicorn
  6. Streaming con tipos seguros (solo REST API)

El generador configura registro estructurado usando AWS Lambda Powertools. Puedes acceder al logger en tus manejadores de rutas:

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}

El logger incluye automáticamente:

  • IDs de correlación para trazado de solicitudes
  • Ruta y método de la solicitud
  • Información del contexto de Lambda
  • Indicadores de inicio en frío

El trazado con AWS X-Ray se configura automáticamente. Puedes añadir subsegmentos personalizados a tus trazas:

from .init import app, tracer
@app.get("/items/{item_id}")
@tracer.capture_method
def read_item(item_id: int):
# Crea un nuevo subsegmento
with tracer.provider.in_subsegment("fetch-item-details"):
# Tu lógica aquí
return {"item_id": item_id}

Las métricas de CloudWatch se recopilan automáticamente para cada solicitud. Puedes añadir métricas personalizadas:

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}

Las métricas por defecto incluyen:

  • Conteo de solicitudes
  • Conteos de éxito/fallo
  • Métricas de inicio en frío
  • Métricas por ruta

El generador incluye manejo de errores completo:

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}

Las excepciones no manejadas son capturadas por el middleware y:

  1. Registran la excepción completa con stack trace
  2. Registran una métrica de fallo
  3. Devuelven una respuesta segura 500 al cliente
  4. Preservan el ID de correlación

Cuando tu API está protegida por autenticación, tus manejadores de rutas a menudo necesitan saber quién está llamando. La FastAPI generada se ejecuta dentro de AWS Lambda mediante el Lambda Web Adapter, que reenvía el contexto de solicitud de API Gateway como JSON en el encabezado x-amzn-request-context. Puedes leerlo desde el Request de FastAPI para extraer la identidad del llamador.

Como ejemplo, agreguemos un endpoint /me que devuelve detalles sobre el usuario que llama. Implementaremos la extracción como una dependencia de FastAPI para que pueda reutilizarse en todas las rutas. La forma del contexto de solicitud — y por lo tanto cómo extraes la identidad — depende tanto de tu método auth seleccionado como de si desplegaste una REST API o HTTP API.

auth = iam

Para autenticación IAM, buscamos al llamador en Cognito usando el sub extraído del contexto de solicitud de API Gateway. Crea identity.py junto a 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:
# El Lambda Web Adapter reenvía el contexto de solicitud de API Gateway como 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(
# Asume que el id del user pool está configurado en el entorno 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

Con auth: 'cognito', el autorizador de Cognito User Pools de API Gateway verifica el JWT que el llamador proporciona en el encabezado Authorization y coloca los claims verificados en el contexto de solicitud.

Crea identity.py junto a 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:
# El Lambda Web Adapter reenvía el contexto de solicitud de API Gateway como 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)]

Luego puedes inyectar la dependencia CurrentUser en cualquier ruta que necesite la identidad del llamador:

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 generada soporta respuestas en streaming de forma predeterminada cuando se usa una REST API. La infraestructura está configurada para usar el AWS Lambda Web Adapter para ejecutar tu FastAPI mediante uvicorn dentro de Lambda, con ResponseTransferMode.STREAM en API Gateway para todas las operaciones de REST API, lo que permite que el streaming funcione junto con operaciones que no son de streaming.

El init.py generado exporta una clase JsonStreamingResponse que proporciona streaming con tipos seguros y generación adecuada del esquema OpenAPI. Esto asegura que el generador connection pueda producir métodos de cliente de streaming correctamente tipados.

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 clase JsonStreamingResponse:

  1. Serializa modelos Pydantic al formato JSON Lines (application/jsonl)
  2. Proporciona un helper openapi_response que genera el esquema OpenAPI correcto con itemSchema, permitiendo que el generador connection produzca métodos de cliente de streaming con tipos seguros

Para consumir un stream de respuestas, puedes usar el generador connection que provee un método tipado para iterar sobre los fragmentos transmitidos.

El generador de FastAPI crea código de infraestructura CDK o Terraform según tu iacProvider seleccionado. Puedes usarlo para desplegar tu API FastAPI.

El constructo CDK para desplegar tu API está en la carpeta common/constructs. Puedes usarlo en una aplicación CDK:

import { MyApi } from ':my-scope/common-constructs';
export class ExampleStack extends Stack {
constructor(scope: Construct, id: string) {
// Añadir la API al stack
const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
});
}
}

Esto configura:

  1. Una función AWS Lambda por cada operación en la aplicación FastAPI
  2. API Gateway HTTP/REST API como trigger de la función
  3. Roles y permisos IAM
  4. Grupo de logs de CloudWatch
  5. Configuración de trazado X-Ray
  6. Namespace de métricas CloudWatch
auth = cognito
auth = custom
infra = rest-lambda

Para las API REST, el constructo generado asocia una Web ACL de AWS WAFv2 con la etapa de API Gateway de forma predeterminada. La Web ACL utiliza el conjunto de reglas predeterminado administrado por AWS (AWSManagedRulesCommonRuleSet y AWSManagedRulesKnownBadInputsRuleSet), proporcionando protección contra exploits web comunes, incluido el OWASP Top 10. Los registros de solicitudes de WAF se escriben en un grupo de CloudWatch Logs.

Puedes editar el constructo rest-api generado para agregar, eliminar o ajustar reglas (por ejemplo, para agregar reglas basadas en tasas o grupos de reglas administradas adicionales).

Para optar por no participar (por ejemplo, para adjuntar tu propia Web ACL), establece enableWaf en false:

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

Para las APIs REST, la infraestructura generada habilita el registro de acceso por defecto, escribiendo una línea JSON estructurada por solicitud en un grupo de CloudWatch Logs dedicado. El grupo de registros está cifrado con una clave KMS administrada por el cliente y se conserva durante un año.

API Gateway escribe registros de acceso utilizando un rol de CloudWatch Logs a nivel de cuenta. Este rol se configura en el ajuste AWS::ApiGateway::Account, que es un singleton por región por cuenta — solo hay un rol para cada API REST en la región. Para gestionar esto de forma segura a través de múltiples stacks desplegados de forma independiente, la infraestructura generada:

  • Crea un rol compartido de CloudWatch Logs y lo configura en la cuenta solo cuando no hay un rol funcional ya establecido, por lo que los despliegues nunca sobrescriben un rol que pertenece a otro stack.
  • Deja el ajuste de la cuenta intacto durante el desmontaje, por lo que destruir un stack nunca deshabilita el registro para otras APIs REST en la región.

El rol de cuenta es gestionado por el constructo ApiGatewayAccount, un singleton con ámbito de stack resuelto a través de ApiGatewayAccount.ensure(scope). Cada etapa de la API REST depende de él, y el rol se configura mediante un recurso personalizado respaldado por Lambda.

Puedes personalizar el formato del registro de acceso pasando deployOptions al construir tu API:

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

Los constructos CDK de la API REST/HTTP están configurados para proporcionar una interfaz type-safe que define integraciones para cada una de tus operaciones.

Los constructos CDK proporcionan soporte completo de integración con seguridad de tipos como se describe a continuación.

Puedes usar el método estático defaultIntegrations para utilizar el patrón por defecto, que define una función AWS Lambda individual para cada operación:

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

Puedes acceder a las funciones AWS Lambda subyacentes a través de la propiedad integrations del constructo de la API, de manera type-safe. Por ejemplo, si tu API define una operación llamada sayHello y necesitas agregar permisos a esta función, puedes hacerlo así:

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
});
// sayHello está tipado según las operaciones definidas en tu API
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({
effect: Effect.ALLOW,
actions: [...],
resources: [...],
}));

Si tu API usa el patrón shared, el router Lambda compartido se expone como api.integrations.$router:

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

Si deseas personalizar las opciones usadas al crear la función Lambda para cada integración por defecto, puedes usar el método withDefaultOptions. Por ejemplo, si quieres que todas tus funciones Lambda residan en una VPC:

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

Para personalizar las opciones usadas al crear la integración por defecto para operaciones específicas (sin afectar a las demás), puedes usar el método withOperationOptions. Por ejemplo, si deseas aumentar el timeout de la función Lambda para solo una operación:

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this)
.withOperationOptions({
sayHello: {
timeout: Duration.seconds(60),
},
})
.build(),
});
// Las operaciones seleccionadas siguen siendo integraciones por defecto, por lo que siguen tipadas en consecuencia:
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({ ... }));

Las opciones que especificas se fusionan con las opciones de integración por defecto (y cualquier opción establecida mediante withDefaultOptions). Ten en cuenta que no puedes especificar opciones para operaciones que hayas reemplazado mediante withOverrides, ya que estas ya no usan la integración por defecto.

Encontrarás un error de tipo si la misma operación es objetivo tanto de withOperationOptions como de withOverrides, independientemente del orden en que los llames.

También puedes sobrescribir integraciones para operaciones específicas usando el método withOverrides. Cada override debe especificar una propiedad integration que está tipada al constructo de integración CDK apropiado para la API HTTP o REST. El método withOverrides también es type-safe. Por ejemplo, si quieres sobrescribir una API getDocumentation para apuntar a documentación alojada en un sitio externo:

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

Notarás que la integración sobrescrita ya no tiene una propiedad handler cuando se accede a través de api.integrations.getDocumentation.

Puedes agregar propiedades adicionales a una integración que también estarán tipadas, permitiendo abstraer otros tipos de integración manteniendo el type-safe. Por ejemplo, si creaste una integración S3 para una API REST y luego quieres referenciar el bucket para una operación particular:

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(),
});
// Más tarde, quizás en otro archivo, puedes acceder a la propiedad bucket que definimos
// de manera type-safe
api.integrations.getFile.bucket.grantRead(...);

También puedes proveer options en tu integración para sobrescribir opciones de método específicas como autorizadores. Por ejemplo, si deseas usar autenticación con Cognito para tu operación getDocumentation:

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

Si prefieres, puedes no usar las integraciones por defecto y proveer directamente una para cada operación. Esto es útil si, por ejemplo, cada operación necesita usar un tipo diferente de integración o quieres recibir un error de tipo al agregar nuevas operaciones:

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

Los constructos de API CDK generados admiten dos patrones de integración:

  • isolated crea una función Lambda por operación. Este es el valor por defecto para las APIs generadas.
  • shared crea un único router Lambda por defecto y lo reutiliza para cada operación a menos que sobrescribas integraciones específicas.

isolated te brinda permisos y configuración más granulares por operación. shared reduce la proliferación de funciones Lambda e integraciones de API Gateway mientras aún permite sobrescritas selectivas.

Por ejemplo, establecer pattern a 'shared' crea una única función en lugar de una por integración:

packages/common/constructs/src/app/apis/my-api.ts
export class MyApi<...> extends ... {
public static defaultIntegrations = (scope: Construct) => {
...
return IntegrationBuilder.rest({
pattern: 'shared',
...
});
};
}

Dado que las operaciones en FastAPI se definen en Python y la infraestructura CDK en TypeScript, instrumentamos generación de código para proveer metadatos al constructo CDK y proporcionar una interfaz tipada para las integraciones.

Se añade un target generate:<ApiName>-metadata al project.json de los constructos comunes para facilitar esta generación de código, que emite un archivo como packages/common/constructs/src/generated/my-api/metadata.gen.ts. Dado que esto se genera en tiempo de build, se ignora en control de versiones.

auth = iam

Si seleccionaste autenticación IAM, puedes usar el método grantInvokeAccess para conceder acceso a tu API:

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

El generador configura un servidor de desarrollo local que puedes ejecutar con:

Terminal window
pnpm nx serve my-api

Esto inicia un servidor de desarrollo FastAPI local con:

  • Recarga automática al cambiar el código
  • Documentación interactiva de la API en /docs o /redoc
  • Esquema OpenAPI en /openapi.json

Para invocar tu API desde un sitio React, puedes usar el generador connection.

Usa el generador connection para integrar este proyecto con otros en tu workspace. Las siguientes conexiones involucran este proyecto:

FastAPI
React a FastAPILlama a una API FastAPI en Python desde un sitio web React
FastAPIAmazon DynamoDBPython
FastAPI a Python DynamoDBConecta una FastAPI a una tabla DynamoDB