FastAPI

FastAPI es un framework para construir APIs en Python.

El generador de FastAPI crea una nueva aplicación FastAPI con configuración de infraestructura en AWS CDK. El backend generado utiliza AWS Lambda para despliegue sin servidor, expuesto a través de un API Gateway de AWS. Configura AWS Lambda Powertools para observabilidad, incluyendo logging, trazado con AWS X-Ray y métricas de Cloudwatch.

Uso

Generar una FastAPI

Puedes generar una nueva FastAPI de dos formas:

VSCode

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#fast-api
5. Complete los parámetros requeridos
6. Haga clic en 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

También puede realizar una ejecución en seco para ver qué archivos se cambiarían

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

Opciones

Parámetro	Tipo	Predeterminado	Descripción
name Requerido	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

Nota

Puedes elegir entre una REST API o una HTTP API para implementar tu API. Consulta la Guía para desarrolladores de API Gateway para ayudarte a decidir.

Salida del generador

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
• Directory<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
• Directoryscripts

  • generate_open_api.py Script para generar esquema OpenAPI desde la app FastAPI

El generador también crea constructs CDK para desplegar tu API, ubicados en el directorio packages/common/constructs.

Implementando tu FastAPI

La implementación principal de la API está en main.py. Aquí defines las rutas y sus implementaciones. Ejemplo:

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

El generador configura automáticamente:

1. Integración con AWS Lambda Powertools para observabilidad
2. Middleware de manejo de errores
3. Correlación de solicitudes/respuestas
4. Recolección de métricas
5. Handler de AWS Lambda usando Mangum

Observabilidad con AWS Lambda Powertools

Logging

Configura logging estructurado usando AWS Lambda Powertools. Accede al logger en tus 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}

El logger incluye automáticamente:

• IDs de correlación para tracing
• Ruta y método de la solicitud
• Información del contexto Lambda
• Indicadores de cold start

Tracing

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

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}

Métricas

Se recogen métricas de CloudWatch automáticamente. Añade 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}

Métricas por defecto incluyen:

• Conteo de solicitudes
• Conteos de éxito/fallo
• Métricas de cold start
• Métricas por ruta

Manejo de errores

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}

Excepciones no capturadas son manejadas por el middleware para:

1. Registrar la excepción completa con stack trace
2. Registrar métrica de fallo
3. Devolver respuesta 500 segura al cliente
4. Preservar el ID de correlación

Consejo

Se recomienda especificar modelos de respuesta para mejor generación de código si usas el generador api-connection. Más detalles aquí.

Streaming

Con FastAPI, puedes transmitir respuestas usando StreamingResponse.

Cambios en infraestructura

Como API Gateway de AWS no soporta streaming, necesitarás desplegar en una plataforma que lo permita. La opción más simple es usar Function URL de AWS Lambda. Para esto, reemplaza el constructo generado common/constructs/src/app/apis/<name>-api.ts con uno que despliegue una Function URL.

Ejemplo de constructo FunctionURL para 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 });

    // Registrar URL en configuración para descubrimiento de clientes
    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',
        },
      },
    });
  }
}

Nota

Para un ejemplo completo, consulta el Tutorial de Dungeon Adventure

Implementación

Una vez actualizada la infraestructura, puedes implementar streaming en FastAPI. La API debe:

• Devolver un StreamingResponse
• Declarar el tipo de cada fragmento de respuesta
• Añadir la extensión OpenAPI x-streaming: true si usarás API Connection.

Ejemplo para transmitir objetos 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")

Consumo

Para consumir streams, usa el Generador de API Connection que provee métodos tipados para iterar fragmentos.

Desplegando tu FastAPI

El generador crea un constructo CDK en common/constructs. Úsalo en una aplicación CDK:

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

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

Esto configura:

1. Función Lambda por operación
2. API Gateway HTTP/REST como trigger
3. Roles y permisos IAM
4. Grupo de logs en CloudWatch
5. Configuración de trazado X-Ray
6. Namespace de métricas en CloudWatch

Nota

Si usas autenticación Cognito, debes proveer la propiedad identity:

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

El constructo UserIdentity se genera con el generador ts#cloudscape-website-auth

Integraciones Tipadas

Los constructos CDK de la API REST/HTTP están configurados para proporcionar una interfaz tipada que permite definir integraciones para cada una de tus operaciones.

Integraciones predeterminadas

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

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

Accediendo a las integraciones

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

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

Personalizando opciones predeterminadas

Si deseas personalizar las opciones utilizadas al crear la función Lambda para cada integración predeterminada, 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(),
});

Sobrescribiendo integraciones

También puedes sobrescribir integraciones para operaciones específicas usando el método withOverrides. Cada sobrescritura 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 está tipado. Por ejemplo, si deseas sobrescribir una API getDocumentation para apuntar a documentación alojada en un sitio web 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 integraciones manteniendo la seguridad de tipos. Por ejemplo, si has creado una integración S3 para una API REST y luego deseas 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(),
});

// Posteriormente, quizás en otro archivo, puedes acceder a la propiedad bucket que definimos
// de manera tipada
api.integrations.getFile.bucket.grantRead(...);

Sobrescribiendo autorizadores

También puedes proporcionar options en tu integración para sobrescribir opciones de método específicas como autorizadores. Por ejemplo, si deseas usar autenticación de 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(),
});

Integraciones explícitas

Si lo prefieres, puedes optar por no usar las integraciones predeterminadas y en su lugar proporcionar una directamente para cada operación. Esto es útil si, por ejemplo, cada operación necesita usar un tipo diferente de integración o deseas recibir un error de tipo al agregar nuevas operaciones:

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

Patrón de enrutador

Si prefieres desplegar una sola función Lambda para manejar todas las solicitudes de la API, puedes modificar libremente el método defaultIntegrations de tu API para crear 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) => {
    const router = new Function(scope, 'RouterHandler', { ... });
    return IntegrationBuilder.rest({
      ...
      defaultIntegrationOptions: {},
      buildDefaultIntegration: (op) => {
        return {
          // Referencia el mismo manejador Lambda de enrutador en cada integración
          integration: new LambdaIntegration(router),
        };
      },
    });
  };
}

Puedes modificar el código de otras formas si lo prefieres, por ejemplo, podrías definir la función router como parámetro de defaultIntegrations en lugar de construirla dentro del método.

Generación de código

Como las operaciones se definen en Python y la infraestructura en TypeScript, usamos generación de código para proveer metadatos al constructo CDK.

Se añade un target generate:<ApiName>-metadata al project.json de los constructs para generar archivos como packages/common/constructs/src/generated/my-api/metadata.gen.ts. Este archivo se ignora en control de versiones.

Nota

Debes ejecutar un build tras cambios en la API para actualizar los tipos:

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

Consejo

Para desarrollo activo, usa nx watch para regenerar tipos automáticamente:

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"

Otorgando acceso (solo IAM)

Si usas autenticación IAM, otorga acceso con grantInvokeAccess:

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

Desarrollo local

El generador configura un servidor de desarrollo local. Ejecuta:

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

Esto inicia un servidor FastAPI con:

• Auto-recarga ante cambios
• Documentación interactiva en /docs o /redoc
• Esquema OpenAPI en /openapi.json

Invocando tu FastAPI

Para invocar la API desde React, usa el generador api-connection.