Pular para o conteúdo
@aws/nx-plugin
Blog

FastAPI

FastAPI é um framework para construção de APIs em Python.

O gerador FastAPI cria uma nova aplicação FastAPI com configuração de infraestrutura AWS CDK ou Terraform. O backend gerado utiliza AWS Lambda para implantação serverless, exposto via AWS API Gateway. Configura AWS Lambda Powertools para observabilidade, incluindo registro de logs, rastreamento com AWS X-Ray e métricas no CloudWatch.

Uso

Seção intitulada “Uso”

Gerar uma API FastAPI

Seção intitulada “Gerar uma API FastAPI”

Você pode gerar uma nova API FastAPI de duas formas:

  1. Instale o Nx Console VSCode Plugin se ainda não o fez
  2. Abra o console Nx no VSCode
  3. Clique em Generate (UI) na seção "Common Nx Commands"
  4. Procure por @aws/nx-plugin - py#fast-api
  5. Preencha os parâmetros obrigatórios
    • Clique em Generate

    Opções

    Seção intitulada “Opções”
    Parâmetro Tipo Padrão Descrição
    name Obrigatório 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.
    iacProvider string Inherit The preferred IaC provider. By default this is inherited from your initial selection.
    moduleName string - Python module name

    Saída do Gerador

    Seção intitulada “Saída do Gerador”

    O gerador criará a seguinte estrutura de projeto no diretório <directory>/<api-name>:

    • project.json Configuração do projeto e alvos de build
    • pyproject.toml Configuração do projeto Python e dependências
    • Directory<module_name>
      • __init__.py Inicialização do módulo
      • init.py Configura o app FastAPI e middleware do powertools
      • main.py Implementação da API
    • Directoryscripts
      • generate_open_api.py Script para gerar schema OpenAPI a partir do app FastAPI

    Infraestrutura

    Seção intitulada “Infraestrutura”

    Como este gerador fornece infraestrutura como código com base no iacProvider escolhido, ele criará um projeto em packages/common que inclui os constructs CDK ou módulos Terraform relevantes.

    O projeto comum de infraestrutura como código está estruturado da seguinte forma:

    • Directorypackages/common/constructs
      • Directorysrc
        • Directoryapp/ Constructs para infraestrutura específica de um projeto/gerador
          • …
        • Directorycore/ Constructs genéricos reutilizados pelos constructs em app
          • …
        • index.ts Ponto de entrada exportando os constructs de app
      • project.json Metas de build e configuração do projeto

    Para implantar sua API, os seguintes arquivos são gerados:

    • Directorypackages/common/constructs/src
      • Directoryapp
        • Directoryapis
          • <project-name>.ts Construto CDK para implantar sua API
      • Directorycore
        • Directoryapi
          • http-api.ts Construto CDK para implantar uma API HTTP (se você escolheu implantar uma API HTTP)
          • rest-api.ts Construto CDK para implantar uma API REST (se você escolheu implantar uma API REST)
          • utils.ts Utilitários para os construtos da API

    Implementando sua API FastAPI

    Seção intitulada “Implementando sua API FastAPI”

    A implementação principal da API está em main.py. É aqui que você define suas rotas e suas implementações. Exemplo:

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

    O gerador configura automaticamente vários recursos:

    1. Integração com AWS Lambda Powertools para observabilidade
    2. Middleware de tratamento de erros
    3. Correlação de requisições/respostas
    4. Coleta de métricas
    5. Handler AWS Lambda usando Mangum

    Observabilidade com AWS Lambda Powertools

    Seção intitulada “Observabilidade com AWS Lambda Powertools”

    Logs

    Seção intitulada “Logs”

    O gerador configura logging estruturado usando AWS Lambda Powertools. Você pode acessar o logger nos 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}

    O logger inclui automaticamente:

    • IDs de correlação para rastreamento
    • Caminho e método da requisição
    • Informações de contexto do Lambda
    • Indicadores de cold start

    Rastreamento

    Seção intitulada “Rastreamento”

    O rastreamento com AWS X-Ray é configurado automaticamente. Você pode adicionar 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ção intitulada “Métricas”

    Métricas no CloudWatch são coletadas automaticamente. Você pode adicionar 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 padrão incluem:

    • Contagem de requisições
    • Contagem de sucessos/falhas
    • Métricas de cold start
    • Métricas por rota

    Tratamento de Erros

    Seção intitulada “Tratamento de Erros”

    O gerador inclui tratamento abrangente de erros:

    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}

    Exceções não tratadas são capturadas pelo middleware e:

    1. Registram a exceção completa com stack trace
    2. Gravam métrica de falha
    3. Retornam resposta 500 segura ao cliente
    4. Preservam o ID de correlação

    Recomenda-se especificar modelos de resposta para operações da API para melhor geração de código ao usar o api-connection. Veja detalhes aqui.

    Streaming

    Seção intitulada “Streaming”

    Com FastAPI, você pode transmitir respostas usando StreamingResponse.

    Alterações na Infraestrutura

    Seção intitulada “Alterações na Infraestrutura”

    Como o AWS API Gateway não suporta respostas em streaming, você precisará implantar sua API em uma plataforma que suporte isso. A opção mais simples é usar Lambda Function URL.

    Para isso, substitua o construct gerado em common/constructs/src/app/apis/<name>-api.ts por um que implemente Function URL.

    Exemplo de Construct com Function URL 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-x86',
                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',
            },
          },
        });
      }
    }

    Implementação

    Seção intitulada “Implementação”

    Após atualizar a infraestrutura, você pode implementar streaming no FastAPI. A API deve:

    Exemplo de streaming de 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

    Seção intitulada “Consumo”

    Para consumir streams, utilize o Gerador de Conexão de API.

    Implantando sua API FastAPI

    Seção intitulada “Implantando sua API FastAPI”

    O gerador cria código de infraestrutura CDK/Terraform baseado no iacProvider. Use para implantar:

    O construct CDK para implantação está em common/constructs. Use em uma aplicação 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(),
        });
      }
    }

    Isso configura:

    1. Função Lambda para cada operação
    2. API Gateway HTTP/REST
    3. Permissões IAM
    4. Log group no CloudWatch
    5. Configuração de X-Ray
    6. Namespace de métricas

    Integrações

    Seção intitulada “Integrações”

    Os construtos CDK da API REST/HTTP são configurados para fornecer uma interface type-safe para definir integrações para cada uma de suas operações.

    Os construtos CDK fornecem suporte completo a integrações type-safe conforme descrito abaixo.

    Integrações Padrão

    Seção intitulada “Integrações Padrão”

    Você pode usar o método estático defaultIntegrations para utilizar o padrão padrão, que define uma função AWS Lambda individual para cada operação:

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

    Acessando Integrações

    Seção intitulada “Acessando Integrações”

    Você pode acessar as funções AWS Lambda subjacentes através da propriedade integrations do construto da API, de forma type-safe. Por exemplo, se sua API define uma operação chamada sayHello e você precisa adicionar permissões a esta função, você pode fazer isso da seguinte forma:

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

    // sayHello é tipado conforme as operações definidas em sua API
    api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({
      effect: Effect.ALLOW,
      actions: [...],
      resources: [...],
    }));

    Personalizando Opções Padrão

    Seção intitulada “Personalizando Opções Padrão”

    Se você deseja personalizar as opções usadas ao criar a função Lambda para cada integração padrão, pode usar o método withDefaultOptions. Por exemplo, se deseja que todas suas funções Lambda residam em uma VPC:

    const vpc = new Vpc(this, 'Vpc', ...);
    

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

    Sobrescrevendo Integrações

    Seção intitulada “Sobrescrevendo Integrações”

    Você também pode sobrescrever integrações para operações específicas usando o método withOverrides. Cada sobrescrita deve especificar uma propriedade integration que é tipada ao construto de integração CDK apropriado para a API HTTP ou REST. O método withOverrides também é type-safe. Por exemplo, se você deseja sobrescrever uma API getDocumentation para apontar para documentação hospedada em um site externo:

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

    Você notará que a integração sobrescrita não terá mais uma propriedade handler quando acessada via api.integrations.getDocumentation.

    Você pode adicionar propriedades adicionais a uma integração que também serão tipadas adequadamente, permitindo que outros tipos de integração sejam abstraídos mas permaneçam type-safe. Por exemplo, se você criou uma integração S3 para uma API REST e depois deseja referenciar o bucket para uma operação específica:

    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, talvez em outro arquivo, você pode acessar a propriedade bucket que definimos
    // de forma type-safe
    api.integrations.getFile.bucket.grantRead(...);

    Sobrescrevendo Autorizadores

    Seção intitulada “Sobrescrevendo Autorizadores”

    Você também pode fornecer options em sua integração para sobrescrever opções específicas de método como autorizadores. Por exemplo, se desejar usar autenticação Cognito para sua operação getDocumentation:

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

    Integrações Explícitas

    Seção intitulada “Integrações Explícitas”

    Se preferir, você pode optar por não usar as integrações padrão e fornecer diretamente uma para cada operação. Isso é útil se, por exemplo, cada operação precisar usar um tipo diferente de integração ou se você quiser receber um erro de tipo ao adicionar novas operações:

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

    Padrão de Roteador

    Seção intitulada “Padrão de Roteador”

    Se você preferir implantar uma única função Lambda para atender todas as requisições da API, pode livremente editar o método defaultIntegrations de sua API para criar uma única função em vez de uma por integração:

    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 o mesmo router lambda handler em todas as integrações
              integration: new LambdaIntegration(router),
            };
          },
        });
      };
    }

    Você pode modificar o código de outras formas se preferir, por exemplo pode definir a função router como parâmetro para defaultIntegrations em vez de construí-la dentro do método.

    Geração de Código

    Seção intitulada “Geração de Código”

    Como as operações são definidas em Python e a infraestrutura em TypeScript, geramos código para fornecer metadados ao construct CDK.

    Um alvo generate:<ApiName>-metadata é adicionado ao project.json para gerar arquivos como packages/common/constructs/src/generated/my-api/metadata.gen.ts.

    Use nx watch para regenerar tipos automaticamente:

    Terminal window
    pnpm nx watch --projects=<FastAPIProject> -- \
    pnpm nx run <InfraProject>:"generate:<ApiName>-metadata"

    Concedendo Acesso (Somente IAM)

    Seção intitulada “Concedendo Acesso (Somente IAM)”

    Para autenticação IAM, use grantInvokeAccess:

    api.grantInvokeAccess(myIdentityPool.authenticatedRole);

    Desenvolvimento Local

    Seção intitulada “Desenvolvimento Local”

    Execute o servidor local com:

    Terminal window
    pnpm nx run my-api:serve

    Isso inicia um servidor FastAPI com:

    • Recarregamento automático
    • Documentação interativa em /docs ou /redoc
    • Schema OpenAPI em /openapi.json

    Invocando sua API FastAPI

    Seção intitulada “Invocando sua API FastAPI”

    Para consumir a API de um site React, use o api-connection.