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. O backend gerado utiliza AWS Lambda para implantação serverless, exposto via AWS API Gateway HTTP API. Configura automaticamente AWS Lambda Powertools para observabilidade, incluindo logging, rastreamento com AWS X-Ray e métricas no CloudWatch.

Utilização

Gerar uma aplicação FastAPI

Você pode gerar uma nova aplicação FastAPI de duas formas:

VSCode

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
6. Clique em 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

Você também pode realizar uma execução simulada para ver quais arquivos seriam alterados

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

Opções

Parâmetro	Tipo	Padrão	Descrição
name Obrigatório	string	-	project name.
directory	string	packages	The directory to store the application in.

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 targets 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 a aplicação FastAPI e middleware do powertools
  • main.py Implementação da API

O gerador também cria constructs CDK para implantação da API, localizados no diretório packages/common/constructs.

Implementando sua FastAPI

A implementação principal da API está em main.py. Aqui você define as 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:

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

Logging

Logging estruturado é configurado automaticamente. 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
• Método e caminho da requisição
• Informações de contexto do Lambda
• Indicadores de cold start

Tracing

Rastreamento com AWS X-Ray é configurado automaticamente. Você pode adicionar subsegmentos:

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

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
• Sucessos/falhas
• Métricas de cold start
• Métricas por rota

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
4. Preservam o ID de correlação

Dica

Recomenda-se especificar modelos de resposta para melhor geração de código ao usar o gerador api-connection. Mais detalhes aqui.

Streaming

Com FastAPI, você pode transmitir respostas usando StreamingResponse.

Alterações na Infraestrutura

Como o API Gateway não suporta streaming, você precisará implantar em uma plataforma compatível. A opção mais simples é usar AWS Lambda Function URL. Para isso, ajuste o construct HttpApi gerado:

Exemplo de Alterações

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

Após as alterações, atualize packages/common/constructs/src/app/http-apis/<my-api>.ts para usar a nova opção de function URL.

Implementação

Com a infraestrutura atualizada, você pode implementar streaming:

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, utilize o Gerador API Connection para iterar de forma type-safe.

Implantação da FastAPI

O construct CDK gerado pode ser usado em aplicações CDK:

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

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    const api = new MyApi(this, 'MyApi');
  }
}

Isso configura:

1. Função Lambda executando sua aplicação
2. API Gateway HTTP API como trigger
3. Permissões IAM
4. Log group no CloudWatch
5. Configuração de tracing X-Ray
6. Namespace de métricas no CloudWatch

Concedendo Acesso

Use grantInvokeAccess para conceder acesso:

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

Desenvolvimento Local

Execute o servidor local com:

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

O servidor inclui:

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

Invocando a API

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