Saltearse al contenido
@aws/nx-plugin
Blog

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 despliegues serverless, 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 en Cloudwatch.

Uso

Generar una API FastAPI

Puedes generar una nueva API FastAPI de dos formas:

  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
    • Haga clic en Generate

    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.

    Resultado del generador

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

    • project.json Configuración del proyecto y targets de build
    • pyproject.toml Configuración de proyecto Python y dependencias
    • Directory<module_name>
      • __init__.py Inicialización del módulo
      • init.py Configura la app FastAPI y 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 solicitud/respuesta
    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 tracing 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):
        with tracer.provider.in_subsegment("fetch-item-details"):
            return {"item_id": item_id}

    Métricas

    Se recopilan métricas en 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 predeterminadas 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. Retornar respuesta 500 segura al cliente
    4. Preservar el ID de correlación

    Se recomienda especificar modelos de respuesta para las operaciones de la API 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 use 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 });
    

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

    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:

    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 Conexión API que provee métodos type-safe para iterar chunks.

    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) {
        const api = new MyApi(this, 'MyApi', {
          integrations: MyApi.defaultIntegrations(this).build(),
        });
      }
    }

    Esto configura:

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

    Si usas autenticación con 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 Type-Safe

    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 metadata type-safe al constructo CDK.

    El target generate:<ApiName>-metadata en project.json genera archivos como packages/common/constructs/src/generated/my-api/metadata.gen.ts. Este archivo se ignora en control de versiones.

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

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

    Usa nx watch para regenerar tipos automáticamente durante desarrollo:

    Terminal window
    pnpm nx watch --projects=<FastAPIProject> -- \
    pnpm 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:

    Terminal window
    pnpm 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 un sitio React, usa el generador api-connection.