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 API Gateway da AWS. Configura o AWS Lambda Powertools para observabilidade, incluindo logging, rastreamento com AWS X-Ray e métricas no Cloudwatch.

Uso

Gerar uma API FastAPI

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

VSCode
CLI

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

pnpm nx g @aws/nx-plugin:py#fast-api

yarn nx g @aws/nx-plugin:py#fast-api

npx nx g @aws/nx-plugin:py#fast-api

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 nx g @aws/nx-plugin:py#fast-api --dry-run

yarn nx g @aws/nx-plugin:py#fast-api --dry-run

npx nx g @aws/nx-plugin:py#fast-api --dry-run

bunx nx g @aws/nx-plugin:py#fast-api --dry-run

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.
moduleName	string	-	Python module name

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 o app FastAPI e middleware powertools
- main.py Implementação da API
Directoryscripts
- generate_open_api.py Script para gerar schema OpenAPI a partir do app FastAPI

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

Implementando sua API 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:

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

Observabilidade com AWS Lambda Powertools

Logging

Configura logging estruturado usando AWS Lambda Powertools. Acesso ao 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
Path e método da requisição
Informações de contexto Lambda
Indicadores de cold start

Rastreamento

Rastreamento com AWS X-Ray configurado automaticamente. Adicione 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

Métricas no CloudWatch coletadas automaticamente. Adicione 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:

Registram exceção completa com stack trace
Gravam métrica de falha
Retornam resposta 500 segura
Preservam ID de correlação

Streaming

Com FastAPI, você pode transmitir respostas usando StreamingResponse.

Alterações de Infraestrutura

Como o API Gateway da AWS não suporta streaming, você precisará implantar em outra plataforma. A opção mais simples é usar Function URL do AWS Lambda. Substitua o construct gerado common/constructs/src/app/apis/<name>-api.ts por um que use Function URL.

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

Implementação

Após atualizar a infraestrutura, implemente streaming no FastAPI:

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, use o Gerador de Conexão de API.

Implantando sua API FastAPI

O gerador cria um construct CDK em common/constructs. Use 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', {
      integrations: MyApi.defaultIntegrations(this).build(),
    });
  }
}

Isso configura:

Função AWS Lambda para cada operação
API Gateway HTTP/REST como trigger
Roles e permissões IAM
Log group no CloudWatch
Configuração X-Ray
Namespace de métricas CloudWatch

Para autenticação Cognito, forneça a propriedade 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,
    });
  }
}

O construct UserIdentity pode ser gerado com o gerador ts#cloudscape-website-auth

Integrações Type-Safe

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

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

Você pode acessar as funções AWS Lambda subjacentes através da propriedade integrations do construto da API, de maneira tipada. Por exemplo, se sua API define uma operação chamada sayHello e você precisa adicionar algumas 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 para 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 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(),
});

Substituindo Integrações

Você também pode substituir integrações para operações específicas usando o método withOverrides. Cada substituição deve especificar uma propriedade integration tipada para o construto de integração CDK apropriado para a API HTTP ou REST. O método withOverrides também é tipado. Por exemplo, se você deseja substituir 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 substituída não terá mais uma propriedade handler quando acessada via api.integrations.getDocumentation.

Você pode adicionar propriedades extras a uma integração que também serão tipadas adequadamente, permitindo que outros tipos de integração sejam abstraídos mas mantenham a tipagem. Por exemplo, se você criou uma integração S3 para uma API REST e posteriormente 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 maneira tipada
api.integrations.getFile.bucket.grantRead(...);

Substituindo Autorizadores

Você também pode fornecer options em sua integração para substituir opções específicas de método, como autorizadores. Por exemplo, para usar autenticação Cognito na 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 preferir, você pode optar por não usar as integrações padrão e fornecer uma integração explicitamente para cada operação. Isso é útil se, por exemplo, cada operação precisar usar um tipo diferente de integração:

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

Padrão de Roteador

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

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

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

Geração de Código

Um target generate:<ApiName>-metadata é adicionado ao project.json para gerar metadados type-safe. Execute builds após alterações na API:

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

yarn nx run-many --target build --all

npx nx run-many --target build --all

bunx nx run-many --target build --all

Use nx watch para regenerar tipos automaticamente:

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

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

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

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

Concessão de Acesso (IAM)

Para autenticação IAM, use grantInvokeAccess:

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

Desenvolvimento Local

Execute o servidor local com:

pnpm nx run my-api:serve

yarn nx run my-api:serve

npx nx run my-api:serve

bunx 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

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