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 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 logging, trazado de AWS X-Ray y métricas de Cloudwatch.

Puedes generar una nueva FastAPI de dos maneras:

Terminal window
pnpm nx g @aws/nx-plugin:py#api --framework=fastapi
También puede realizar una ejecución en seco para ver qué archivos se cambiarían
Terminal window
pnpm nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run
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 construcción
  • pyproject.toml Configuración del proyecto Python y dependencias
  • run.sh Script de arranque de Lambda Web Adapter para iniciar la aplicación FastAPI a través de uvicorn
  • Directorio<module_name>
    • __init__.py Inicialización del módulo
    • init.py Configura la aplicación FastAPI y configura el middleware de powertools
    • main.py Implementación de la API
  • Directorioscripts
    • generate_open_api.py Script para generar un esquema OpenAPI desde la aplicación FastAPI

Dado que este generador proporciona infraestructura como código basada en tu iac elegido, creará un proyecto en packages/common que incluye las construcciones CDK o módulos Terraform relevantes.

El proyecto común de infraestructura como código está estructurado de la siguiente manera:

  • Directoriopackages/common/constructs
    • Directoriosrc
      • Directorioapp/ Constructs for infrastructure specific to a project/generator
      • Directoriocore/ Generic constructs which are reused by constructs in app
      • index.ts Entry point exporting constructs from app
    • project.json Project build targets and configuration

Para desplegar tu API, se generan los siguientes archivos:

  • Directoriopackages/common/constructs/src
    • Directorioapp
      • Directorioapis
        • <project-name>.ts CDK construct for deploying your API
    • Directoriocore
      • Directorioapi
        • http-api.ts CDK construct for deploying an HTTP API (if you selected to deploy an HTTP API)
        • rest-api.ts CDK construct for deploying a REST API (if you selected to deploy a REST API)
        • utils.ts Utilities for the API constructs

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 a la etapa 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 tus rutas de API y sus implementaciones. Aquí hay 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 varias características automáticamente:

  1. Integración de AWS Lambda Powertools para observabilidad
  2. Middleware de manejo de errores
  3. Correlación de solicitud/respuesta
  4. Recopilación de métricas
  5. Despliegue de AWS Lambda a través de Lambda Web Adapter con uvicorn
  6. Streaming con seguridad de tipos (solo API REST)

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

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 el rastreo de solicitudes
  • Ruta de solicitud, ruta coincidente y método

El trazado de AWS X-Ray se configura automáticamente. Puedes agregar subsegmentos personalizados a tus trazas:

from .init import app, tracer
@app.get("/items/{item_id}")
@tracer.capture_method
def read_item(item_id: int):
# Creates a new subsegment
with tracer.provider.in_subsegment("fetch-item-details"):
# Your logic here
return {"item_id": item_id}

Las métricas de CloudWatch se recopilan automáticamente para cada solicitud. Puedes agregar 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 predeterminadas incluyen:

  • Conteos de solicitudes
  • Conteos de éxito/fallo
  • Métricas por ruta (a través de una dimensión route de <method> <path>)

El generador incluye manejo de errores integral:

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 el stack trace
  2. Registran una métrica de fallo
  3. Devuelven una respuesta 500 segura al cliente
  4. Preservan el ID de correlación

Cuando tu API está protegida por autenticación, tus manejadores de ruta a menudo necesitan saber quién está llamando. La FastAPI generada se ejecuta dentro de AWS Lambda a través del 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 devuelva 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 API REST o HTTP.

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:
# The Lambda Web Adapter forwards the API Gateway request context as 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(
# Assumes user pool id is configured in lambda environment
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 las reclamaciones verificadas 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:
# The Lambda Web Adapter forwards the API Gateway request context as 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 admite respuestas de streaming de forma predeterminada cuando se usa una API REST. La infraestructura está configurada para usar el AWS Lambda Web Adapter para ejecutar tu FastAPI a través de uvicorn dentro de Lambda, con ResponseTransferMode.STREAM en API Gateway para todas las operaciones de API REST, 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 seguridad de tipos y generación adecuada de 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 seguridad de tipos

Para consumir un flujo de respuestas, puedes hacer uso del generador connection que proporcionará un método con seguridad de tipos para iterar sobre tus fragmentos transmitidos.

El generador de FastAPI crea infraestructura como código CDK o Terraform basada en tu iac seleccionado. Puedes usar esto para desplegar tu FastAPI.

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

import { MyApi } from '@my-scope/common-constructs';
export class ExampleStack extends Stack {
constructor(scope: Construct, id: string) {
// Add the api to your stack
const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
});
}
}

Esto configura:

  1. Una función AWS Lambda para cada operación en la aplicación FastAPI
  2. API Gateway HTTP/REST API como disparador de la función
  3. Roles y permisos de IAM
  4. Grupo de logs de CloudWatch
  5. Configuración de trazado de X-Ray
  6. Espacio de nombres de métricas de 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, incluidos los 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 tasa 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 REST APIs, la infraestructura generada habilita el registro de acceso de forma predeterminada, escribiendo una línea JSON estructurada por solicitud en un grupo de CloudWatch Logs dedicado. El grupo de logs está cifrado con una clave KMS administrada por el cliente y se retiene durante un año.

API Gateway escribe los logs de acceso utilizando un rol de CloudWatch Logs a nivel de cuenta. Este rol se configura en la configuración AWS::ApiGateway::Account, que es un singleton por región por cuenta: solo hay un rol para cada REST API en la región. Para gestionar esto de forma segura en múltiples stacks implementados 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 las implementaciones nunca sobrescriben un rol que pertenece a otro stack.
  • Deja la configuración de la cuenta intacta durante el desmontaje, por lo que destruir un stack nunca deshabilita el registro para otras REST APIs en la región.

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

El formato del log de acceso es establecido por el constructo RestApi que tu API extiende. Para personalizarlo, pasa deployOptions a super en el archivo generado packages/common/constructs/src/app/apis/my-api.ts, manteniendo el tracingEnabled que el constructo ya establece:

packages/common/constructs/src/app/apis/my-api.ts
super(scope, id, {
apiName: 'MyApi',
// ...
deployOptions: {
tracingEnabled: true,
accessLogFormat: AccessLogFormat.clf(),
},
...props,
});

AccessLogFormat se importa desde aws-cdk-lib/aws-apigateway. Cualquier cosa que dejes sin configurar mantiene el valor predeterminado del constructo: un formato JSON con los campos estándar.

Los constructos CDK de API REST/HTTP están configurados para proporcionar una interfaz con seguridad de tipos para definir integraciones para cada una de sus operaciones.

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

Puede usar el método estático defaultIntegrations para hacer uso del patrón predeterminado, que define una función AWS Lambda individual para cada operación:

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

Puede acceder a las funciones AWS Lambda subyacentes a través de la propiedad integrations del constructo de API, de manera segura en cuanto a tipos. Por ejemplo, si su API define una operación llamada sayHello y necesita agregar algunos permisos a esta función, puede hacerlo de la siguiente manera:

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
});
// sayHello is typed to the operations defined in your API
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({
effect: Effect.ALLOW,
actions: [...],
resources: [...],
}));

Si su API utiliza el patrón shared, el enrutador 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');

Personalización de Opciones Predeterminadas

Sección titulada «Personalización de Opciones Predeterminadas»

Si desea personalizar las opciones utilizadas al crear la función Lambda para cada integración predeterminada, puede usar el método withDefaultOptions. Por ejemplo, si desea que todas sus 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 utilizadas para crear la integración predeterminada para operaciones específicas (sin afectar a las demás), puede usar el método withOperationOptions. Por ejemplo, si desea aumentar el tiempo de espera 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(),
});
// The selected operations remain default integrations, so they're still typed accordingly:
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({ ... }));

Las opciones que especifique se fusionan con las opciones de integración predeterminadas (y cualquier opción establecida a través de withDefaultOptions). Tenga en cuenta que no puede especificar opciones para operaciones que haya reemplazado a través de withOverrides, ya que estas ya no usan la integración predeterminada.

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

También puede anular integraciones para operaciones específicas usando el método withOverrides. Cada anulación 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 seguro en cuanto a tipos. Por ejemplo, si desea anular una API getDocumentation para que apunte a documentación alojada por algún sitio web externo, podría lograrlo de la siguiente manera:

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

También notará que la integración anulada ya no tiene una propiedad handler al acceder a ella a través de api.integrations.getDocumentation.

Puede agregar propiedades adicionales a una integración que también se tipará en consecuencia, permitiendo que otros tipos de integración se abstraigan pero permanezcan seguros en cuanto a tipos, por ejemplo, si ha creado una integración S3 para una API REST y luego desea hacer referencia al bucket para una operación en particular, puede hacerlo de la siguiente manera:

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(),
});
// Later, perhaps in another file, you can access the bucket property we defined
// in a type-safe manner
api.integrations.getFile.bucket.grantRead(...);

También puede proporcionar options en su integración para anular opciones de método particulares como autorizadores, por ejemplo, si desea usar autenticación Cognito para su operación getDocumentation:

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

Si lo prefiere, puede optar por no usar las integraciones predeterminadas y en su lugar proporcionar directamente una para cada operación. Esto es útil si, por ejemplo, cada operación necesita usar un tipo diferente de integración o si desea 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 predeterminado para las API generadas.
  • shared crea un único enrutador Lambda predeterminado y lo reutiliza para cada operación a menos que anule integraciones específicas.

isolated le brinda permisos y configuración más detallados por operación. shared reduce la proliferación de Lambda y la integración de API Gateway mientras permite anulaciones selectivas.

Por ejemplo, establecer pattern en '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 la generación de código para proporcionar metadatos al constructo CDK y proporcionar una interfaz con seguridad de tipos para las integraciones.

Se agrega un objetivo 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 compilación, se ignora en el control de versiones.

auth = iam

Si seleccionaste usar 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 local de FastAPI con:

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

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

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

FastAPI
React a FastAPILlamar a una FastAPI de Python desde un sitio web React
FastAPIAmazon DynamoDBPython
FastAPI a Python DynamoDBConectar una FastAPI a una tabla DynamoDB