Saltearse al contenido
@aws/nx-plugin
Blog

FastAPI

FastAPI es un framework para construir APIs en Python.

El generador de FastAPI crea una nueva API 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 HTTP API de AWS. Configura AWS Lambda Powertools para observabilidad, incluyendo logging, trazado con AWS X-Ray y métricas de 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 - project name.
    directory string packages The directory to store the application in.

    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 compilación
    • 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

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

    Implementando tu API FastAPI

    La implementación principal de la API está en main.py. Aquí es donde defines las rutas de tu API 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 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. Manejador AWS Lambda usando Mangum

    Observabilidad con AWS Lambda Powertools

    Logging

    El generador configura logging 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 trazabilidad
    • Ruta y método de la solicitud
    • Información del contexto Lambda
    • Indicadores de cold start

    Trazado

    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

    Las métricas de CloudWatch se recogen 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 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}

    Las excepciones no capturadas son manejadas por el middleware y:

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

    Streaming

    Con FastAPI, puedes transmitir respuestas usando el tipo StreamingResponse.

    Cambios en infraestructura

    Como AWS API Gateway no soporta respuestas en streaming, necesitarás desplegar tu FastAPI en una plataforma que lo soporte. La opción más simple es usar una Function URL de AWS Lambda. Para esto, puedes modificar el constructo generado HttpApi para añadir streaming condicionalmente.

    Ejemplo de cambios
     import { Construct } from 'constructs';
    import { CfnOutput, Duration } from 'aws-cdk-lib';
    import { CfnOutput, Duration, Stack } from 'aws-cdk-lib';
     import {
       CorsHttpMethod,
       HttpApi as _HttpApi,
    @@ -7,7 +7,16 @@ import {
       IHttpRouteAuthorizer,
     } from 'aws-cdk-lib/aws-apigatewayv2';
    

           },
         });
    

        this.api = new _HttpApi(this, id, {
          corsPreflight: {
            allowOrigins: props.allowedOrigins ?? ['*'],
            allowMethods: [CorsHttpMethod.ANY],
            allowHeaders: [
              'authorization',
              'content-type',
              'x-amz-content-sha256',
              'x-amz-date',
              'x-amz-security-token',
            ],
          },
          defaultAuthorizer: props.defaultAuthorizer,
        });
        let apiUrl;
        if (props.apiType === 'api-gateway') {
          this.api = new _HttpApi(this, id, {
            corsPreflight: {
              allowOrigins: props.allowedOrigins ?? ['*'],
              allowMethods: [CorsHttpMethod.ANY],
              allowHeaders: [
                'authorization',
                'content-type',
                'x-amz-content-sha256',
                'x-amz-date',
                'x-amz-security-token',
              ],
            },
            defaultAuthorizer: props.defaultAuthorizer,
          });
    

        this.api.addRoutes({
          path: '/{proxy+}',
          methods: [
            HttpMethod.GET,
            HttpMethod.DELETE,
            HttpMethod.POST,
            HttpMethod.PUT,
            HttpMethod.PATCH,
            HttpMethod.HEAD,
          ],
          integration: new HttpLambdaIntegration(
            'RouterIntegration',
            this.routerFunction,
          ),
        });
          this.api.addRoutes({
            path: '/{proxy+}',
            methods: [
              HttpMethod.GET,
              HttpMethod.DELETE,
              HttpMethod.POST,
              HttpMethod.PUT,
              HttpMethod.PATCH,
              HttpMethod.HEAD,
            ],
            integration: new HttpLambdaIntegration(
              'RouterIntegration',
              this.routerFunction,
            ),
          });
          apiUrl = this.api.url;
        } else {
          const stack = Stack.of(this);
          this.routerFunction.addLayers(
            LayerVersion.fromLayerVersionArn(
              this,
              'LWALayer',
              `arn:aws:lambda:${stack.region}:753240598075:layer:LambdaAdapterLayerX86:24`,
            ),
          );
          this.routerFunction.addEnvironment('PORT', '8000');
          this.routerFunction.addEnvironment(
            'AWS_LWA_INVOKE_MODE',
            'response_stream',
          );
          this.routerFunction.addEnvironment(
            'AWS_LAMBDA_EXEC_WRAPPER',
            '/opt/bootstrap',
          );
          this.routerFunctionUrl = this.routerFunction.addFunctionUrl({
            authType: FunctionUrlAuthType.AWS_IAM,
            invokeMode: InvokeMode.RESPONSE_STREAM,
            cors: {
              allowedOrigins: props.allowedOrigins ?? ['*'],
              allowedHeaders: [
                'authorization',
                'content-type',
                'x-amz-content-sha256',
                'x-amz-date',
                'x-amz-security-token',
              ],
            },
          });
          apiUrl = this.routerFunctionUrl.url;
        }
    

        new CfnOutput(this, `${props.apiName}Url`, { value: this.api.url! });
        new CfnOutput(this, `${props.apiName}Url`, { value: apiUrl! });
    

         RuntimeConfig.ensure(this).config.httpApis = {
           ...RuntimeConfig.ensure(this).config.httpApis!,
          [props.apiName]: this.api.url!,
          [props.apiName]: apiUrl,
         };
       }
    

       public grantInvokeAccess(role: IRole) {
        role.addToPrincipalPolicy(
          new PolicyStatement({
            effect: Effect.ALLOW,
            actions: ['execute-api:Invoke'],
            resources: [this.api.arnForExecuteApi('*', '/*', '*')],
          }),
        );
        if (this.api) {
          role.addToPrincipalPolicy(
            new PolicyStatement({
              effect: Effect.ALLOW,
              actions: ['execute-api:Invoke'],
              resources: [this.api.arnForExecuteApi('*', '/*', '*')],
            }),
          );
        } else if (this.routerFunction) {
          role.addToPrincipalPolicy(
            new PolicyStatement({
              effect: Effect.ALLOW,
              actions: ['lambda:InvokeFunctionUrl'],
              resources: [this.routerFunction.functionArn],
              conditions: {
                StringEquals: {
                  'lambda:FunctionUrlAuthType': 'AWS_IAM',
                },
              },
            }),
          );
        }
       }
     }

    Después de estos cambios, actualiza packages/common/constructs/src/app/http-apis/<my-api>.ts para usar la nueva opción de function url.

    Implementación

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

    • Devolver un StreamingResponse
    • Declarar el tipo de cada chunk de respuesta
    • Añadir la extensión OpenAPI x-streaming: true si planeas usar 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, puedes usar el generador API Connection que provee métodos tipados para iterar chunks.

    Desplegando tu API FastAPI

    El generador crea un constructo CDK en 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) {
        // Agrega la API a tu stack
        const api = new MyApi(this, 'MyApi');
      }
    }

    Esto configura:

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

    Otorgando acceso

    Puedes usar grantInvokeAccess para dar acceso:

    api.grantInvokeAccess(myIdentityPool.authenticatedRole);

    Desarrollo local

    El generador configura un servidor local que puedes iniciar con:

    Terminal window
    pnpm nx run my-api:serve

    Esto inicia un servidor de desarrollo FastAPI con:

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

    Invocando tu API

    Para invocar tu API desde un sitio React, usa el generador api-connection.