Pular para o conteúdo

FastAPI

Filter this guidePick generator option values to hide sections that don't apply.

FastAPI é um framework para construir APIs em Python.

O gerador FastAPI cria um novo FastAPI com configuração de infraestrutura AWS CDK ou Terraform. O backend gerado usa AWS Lambda para implantação serverless, exposto via uma API do AWS API Gateway. Ele configura AWS Lambda Powertools para observabilidade, incluindo logging, rastreamento AWS X-Ray e Métricas do Cloudwatch.

Você pode gerar um novo FastAPI de duas maneiras:

Terminal window
pnpm nx g @aws/nx-plugin:py#api --framework=fastapi
Você também pode realizar uma execução simulada para ver quais arquivos seriam alterados
Terminal window
pnpm nx g @aws/nx-plugin:py#api --framework=fastapi --dry-run
ParâmetroTipoPadrãoDescrição
name Obrigatóriostring-Nome do projeto de API a ser gerado
framework fastapifastapiO framework de API a ser utilizado.
integrationPattern isolated | sharedisolatedComo as integrações do API Gateway são geradas para a API. Escolha entre isolated (padrão) e shared.
auth iam | cognito | customiamO método usado para autenticar com sua API. Escolha entre iam (padrão), cognito ou custom.
directory stringpackagesO diretório para armazenar a aplicação.
subDirectory string-O subdiretório onde o projeto é colocado. Por padrão, este é o nome do projeto.
iac inherit | cdk | terraforminheritO provedor IaC preferido. Por padrão, isso é herdado da sua seleção inicial.
moduleName string-Nome do módulo Python
infra rest-lambda | http-lambda | nonerest-lambdaO tipo de infraestrutura a ser usado para implantar esta API.
preferInstallDependencies booleantrueSe deve preferir instalar dependências após a execução do gerador. Defina como false para adiar a instalação ao executar múltiplos geradores em lote (uma instalação ainda é executada se necessário para que geradores subsequentes possam calcular o grafo de projetos Nx); instale uma vez no final.

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 e dependências do projeto Python
  • run.sh Script de bootstrap do Lambda Web Adapter para iniciar o app FastAPI via uvicorn
  • Directory<module_name>
    • __init__.py Inicialização do módulo
    • init.py Configura o app FastAPI e configura o middleware powertools
    • main.py Implementação da API
  • Directoryscripts
    • generate_open_api.py Script para gerar um esquema OpenAPI a partir do app FastAPI

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

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

  • Directorypackages/common/constructs
    • Directorysrc
      • Directoryapp/ Constructs for infrastructure specific to a project/generator
      • Directorycore/ Generic constructs which are reused by constructs in app
      • index.ts Entry point exporting constructs from app
    • project.json Project build targets and configuration

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

  • Directorypackages/common/constructs/src
    • Directoryapp
      • Directoryapis
        • <project-name>.ts CDK construct for deploying your API
    • Directorycore
      • Directoryapi
        • http-api.ts CDK construct for deploying an HTTP API (if you selected to deploy an HTTP API)
        • rest-api.ts CDK construct for deploying a REST API (if you selected to deploy a REST API)
        • utils.ts Utilities for the API constructs

A aplicação implantada possui a seguinte arquitetura:

ClientWAFAPI Gateway(REST API)LambdaCloudWatch(Logs, Metrics)X-Ray(Traces)

As REST APIs incluem um Web ACL do AWS WAFv2 na frente do estágio do API Gateway com o conjunto de regras padrão gerenciado pela AWS habilitado.

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

from pydantic import BaseModel
from .init import app, tracer
class Item(BaseModel):
name: str
@app.get("/items/{item_id}")
@tracer.capture_method
def get_item(item_id: int) -> Item:
return Item(name=...)
@app.post("/items")
@tracer.capture_method
def create_item(item: Item):
return ...

O gerador configura vários recursos automaticamente:

  1. Integração do AWS Lambda Powertools para observabilidade
  2. Middleware de tratamento de erros
  3. Correlação de solicitação/resposta
  4. Coleta de métricas
  5. Implantação do AWS Lambda via Lambda Web Adapter com uvicorn
  6. Streaming type-safe (somente API REST)

O gerador configura logging estruturado usando AWS Lambda Powertools. Você pode acessar o logger em seus manipuladores de rota:

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 de solicitações
  • Caminho da solicitação, rota correspondente e método

O rastreamento AWS X-Ray é configurado automaticamente. Você pode adicionar subsegmentos personalizados aos seus rastreamentos:

from .init import app, tracer
@app.get("/items/{item_id}")
@tracer.capture_method
def read_item(item_id: int):
# Creates a new subsegment
with tracer.provider.in_subsegment("fetch-item-details"):
# Your logic here
return {"item_id": item_id}

As métricas do CloudWatch são coletadas automaticamente para cada solicitação. 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}

As métricas padrão incluem:

  • Contagens de solicitações
  • Contagens de sucesso/falha
  • Métricas por rota (via uma dimensão route de <method> <path>)

O gerador inclui tratamento de erros abrangente:

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. Registram uma métrica de falha
  3. Retornam uma resposta 500 segura ao cliente
  4. Preservam o ID de correlação

Quando sua API é protegida por autenticação, seus manipuladores de rota frequentemente precisam saber quem está chamando. O FastAPI gerado é executado dentro do AWS Lambda via o Lambda Web Adapter, que encaminha o contexto de solicitação do API Gateway como JSON no cabeçalho x-amzn-request-context. Você pode lê-lo do Request do FastAPI para extrair a identidade do chamador.

Como exemplo, vamos adicionar um endpoint /me que retorna detalhes sobre o usuário chamador. Implementaremos a extração como uma dependência do FastAPI para que possa ser reutilizada entre rotas. A forma do contexto de solicitação — e, portanto, como você extrai a identidade — depende tanto do seu método auth selecionado quanto se você implantou uma API REST ou HTTP.

auth = iam

Para autenticação IAM, procuramos o chamador no Cognito usando o sub extraído do contexto de solicitação do API Gateway. Crie identity.py ao lado de main.py:

import json
import os
from typing import Annotated
from boto3 import client
from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel
cognito = client("cognito-idp")
class Identity(BaseModel):
sub: str
username: str
def get_identity(request: Request) -> Identity:
# The Lambda Web Adapter forwards the API Gateway request context as JSON
request_context_header = request.headers.get("x-amzn-request-context")
if not request_context_header:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
request_context = json.loads(request_context_header)
provider = request_context.get("identity", {}).get("cognitoAuthenticationProvider")
sub = provider.split(":")[-1] if provider else None
if not sub:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
users = cognito.list_users(
# Assumes user pool id is configured in lambda environment
UserPoolId=os.environ["USER_POOL_ID"],
Limit=1,
Filter=f'sub="{sub}"',
).get("Users", [])
if len(users) != 1:
raise HTTPException(status_code=403, detail=f"No user found with subjectId {sub}")
return Identity(sub=sub, username=users[0]["Username"])
CurrentUser = Annotated[Identity, Depends(get_identity)]
auth = cognito

Com auth: 'cognito', o autorizador Cognito User Pools do API Gateway verifica o JWT que o chamador fornece no cabeçalho Authorization e coloca as claims verificadas no contexto de solicitação.

Crie identity.py ao lado de main.py:

import json
from typing import Annotated
from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel
class Identity(BaseModel):
sub: str
username: str
def get_identity(request: Request) -> Identity:
# The Lambda Web Adapter forwards the API Gateway request context as JSON
request_context_header = request.headers.get("x-amzn-request-context")
if not request_context_header:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
request_context = json.loads(request_context_header)
claims = request_context.get("authorizer", {}).get("claims", {})
sub = claims.get("sub")
username = claims.get("username")
if not sub or not username:
raise HTTPException(status_code=403, detail="Unable to determine calling user")
return Identity(sub=sub, username=username)
CurrentUser = Annotated[Identity, Depends(get_identity)]

Você pode então injetar a dependência CurrentUser em qualquer rota que precise da identidade do chamador:

from .identity import CurrentUser, Identity
from .init import app, tracer
@app.get("/me")
@tracer.capture_method
def me(identity: CurrentUser) -> Identity:
return identity
infra = rest-lambda

O FastAPI gerado suporta respostas de streaming prontas para uso ao usar uma API REST. A infraestrutura é configurada para usar o AWS Lambda Web Adapter para executar seu FastAPI via uvicorn dentro do Lambda, com ResponseTransferMode.STREAM no API Gateway para todas as operações da API REST, o que permite que o streaming funcione ao lado de operações sem streaming.

O init.py gerado exporta uma classe JsonStreamingResponse que fornece streaming type-safe com geração adequada de esquema OpenAPI. Isso garante que o gerador connection possa produzir métodos de cliente de streaming corretamente tipados.

from pydantic import BaseModel
from .init import app, JsonStreamingResponse
class Chunk(BaseModel):
message: str
async def generate_chunks():
for i in range(100):
yield Chunk(message=f"This is chunk {i}")
@app.post(
"/stream",
response_class=JsonStreamingResponse,
responses={200: JsonStreamingResponse.openapi_response(Chunk, "Stream of chunks")},
)
async def my_stream() -> JsonStreamingResponse:
return JsonStreamingResponse(generate_chunks())

A classe JsonStreamingResponse:

  1. Serializa modelos Pydantic para o formato JSON Lines (application/jsonl)
  2. Fornece um auxiliar openapi_response que gera o esquema OpenAPI correto com itemSchema, permitindo que o gerador connection produza métodos de cliente de streaming type-safe

Para consumir um fluxo de respostas, você pode fazer uso do gerador connection que fornecerá um método type-safe para iterar sobre seus chunks transmitidos.

O gerador FastAPI cria infraestrutura como código CDK ou Terraform com base no seu iac selecionado. Você pode usar isso para implantar seu FastAPI.

O construto CDK para implantar sua API na pasta common/constructs. Você pode usar isso em uma aplicação CDK:

import { MyApi } from '@my-scope/common-constructs';
export class ExampleStack extends Stack {
constructor(scope: Construct, id: string) {
// Add the api to your stack
const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
});
}
}

Isso configura:

  1. Uma função AWS Lambda para cada operação na aplicação FastAPI
  2. API Gateway HTTP/REST API como gatilho da função
  3. Funções e permissões IAM
  4. Grupo de logs do CloudWatch
  5. Configuração de rastreamento X-Ray
  6. Namespace de métricas do CloudWatch
auth = cognito
auth = custom
infra = rest-lambda

Para APIs REST, o construto gerado associa um Web ACL do AWS WAFv2 com o estágio do API Gateway por padrão. O Web ACL usa o conjunto de regras padrão gerenciado pela AWS (AWSManagedRulesCommonRuleSet e AWSManagedRulesKnownBadInputsRuleSet), fornecendo proteção contra explorações web comuns, incluindo o OWASP Top 10. Os logs de solicitações do WAF são gravados em um grupo do CloudWatch Logs.

Você pode editar o construto rest-api gerado para adicionar, remover ou ajustar regras (por exemplo, para adicionar regras baseadas em taxa ou grupos de regras gerenciadas adicionais).

Para desativar (por exemplo, para anexar seu próprio Web ACL), defina enableWaf como false:

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
enableWaf: false,
});
infra = rest-lambda

Para APIs REST, a infraestrutura gerada habilita o registro de acesso por padrão, escrevendo uma linha JSON estruturada por requisição em um grupo dedicado do CloudWatch Logs. O grupo de logs é criptografado com uma chave KMS gerenciada pelo cliente e retido por um ano.

O API Gateway escreve logs de acesso usando uma função do CloudWatch Logs no nível da conta. Esta função é configurada na configuração AWS::ApiGateway::Account, que é um singleton por região por conta — há apenas uma função para cada API REST na região. Para gerenciar isso com segurança em várias pilhas implantadas independentemente, a infraestrutura gerada:

  • Cria uma função compartilhada do CloudWatch Logs e a configura na conta apenas quando nenhuma função funcional já está definida, para que as implantações nunca sobrescrevam uma função que outra pilha possui.
  • Deixa a configuração da conta intocada na desmontagem, para que destruir uma pilha nunca desabilite o registro para outras APIs REST na região.

A função da conta é gerenciada pelo construto ApiGatewayAccount, um singleton com escopo de pilha resolvido via ApiGatewayAccount.ensure(scope). O estágio de cada API REST depende dele, e a função é configurada por um recurso personalizado baseado em Lambda.

O formato do log de acesso é definido pelo construto RestApi que sua API estende. Para personalizá-lo, passe deployOptions para super no packages/common/constructs/src/app/apis/my-api.ts gerado, mantendo o tracingEnabled que o construto já define:

packages/common/constructs/src/app/apis/my-api.ts
super(scope, id, {
apiName: 'MyApi',
// ...
deployOptions: {
tracingEnabled: true,
accessLogFormat: AccessLogFormat.clf(),
},
...props,
});

AccessLogFormat é importado de aws-cdk-lib/aws-apigateway. Qualquer coisa que você deixar indefinida mantém o padrão do construto — um formato JSON com os campos padrão.

Os construtos CDK de 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.

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

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

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 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 is typed to the operations defined in your API
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({
effect: Effect.ALLOW,
actions: [...],
resources: [...],
}));

Se sua API usa o padrão shared, o Lambda router compartilhado é exposto como api.integrations.$router:

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this).build(),
});
api.integrations.$router.handler.addEnvironment('LOG_LEVEL', 'DEBUG');

Se você deseja personalizar as opções usadas ao criar a função Lambda para cada integração padrão, você pode usar o método withDefaultOptions. Por exemplo, se você deseja que todas as 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(),
});

Para personalizar as opções usadas para criar a integração padrão para operações específicas (sem afetar as outras), você pode usar o método withOperationOptions. Por exemplo, se você deseja aumentar o timeout da função Lambda para apenas uma operação:

const api = new MyApi(this, 'MyApi', {
integrations: MyApi.defaultIntegrations(this)
.withOperationOptions({
sayHello: {
timeout: Duration.seconds(60),
},
})
.build(),
});
// The selected operations remain default integrations, so they're still typed accordingly:
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({ ... }));

As opções que você especifica são mescladas com as opções de integração padrão (e quaisquer opções definidas via withDefaultOptions). Observe que você não pode especificar opções para operações que você substituiu via withOverrides, pois estas não usam mais a integração padrão.

Você encontrará um erro de tipo se a mesma operação for alvo de ambos withOperationOptions e withOverrides, independentemente da ordem em que você os chamar.

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 que é tipada para o 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 substituir uma API getDocumentation para apontar para documentação hospedada por algum site externo, você pode fazer isso da seguinte forma:

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

Você também notará que a integração substituída não tem mais uma propriedade handler ao acessá-la 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, você pode fazer isso da seguinte forma:

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(),
});
// Later, perhaps in another file, you can access the bucket property we defined
// in a type-safe manner
api.integrations.getFile.bucket.grantRead(...);

Você também pode fornecer options em sua integração para substituir opções de método específicas, como autorizadores, por exemplo, se você deseja 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(...) // for REST, or HttpUserPoolAuthorizer for an HTTP API
}
},
})
.build(),
});

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

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

Os construtos de API CDK gerados suportam dois padrões de integração:

  • isolated cria uma função Lambda por operação. Este é o padrão para APIs geradas.
  • shared cria um único Lambda router padrão e o reutiliza para cada operação, a menos que você substitua integrações específicas.

isolated oferece permissões e configuração mais granulares por operação. shared reduz a proliferação de Lambda e integrações do API Gateway, enquanto ainda permite substituições seletivas.

Por exemplo, definir pattern como 'shared' cria 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) => {
...
return IntegrationBuilder.rest({
pattern: 'shared',
...
});
};
}

Como as operações no FastAPI são definidas em Python e a infraestrutura CDK em TypeScript, instrumentamos a geração de código para fornecer metadados ao construto CDK para fornecer uma interface type-safe para integrações.

Um alvo generate:<ApiName>-metadata é adicionado ao project.json dos construtos comuns para facilitar essa geração de código, que emite um arquivo como packages/common/constructs/src/generated/my-api/metadata.gen.ts. Como isso é gerado em tempo de build, é ignorado no controle de versão.

auth = iam

Se você selecionou usar autenticação IAM, você pode usar o método grantInvokeAccess para conceder acesso à sua API:

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

O gerador configura um servidor de desenvolvimento local que você pode executar com:

Terminal window
pnpm nx serve my-api

Isso inicia um servidor de desenvolvimento FastAPI local com:

  • Recarga automática em alterações de código
  • Documentação interativa da API em /docs ou /redoc
  • Esquema OpenAPI em /openapi.json

Para invocar sua API a partir de um site React, você pode usar o gerador connection.

Use o gerador connection para integrar este projeto com outros em seu workspace. As seguintes conexões envolvem este projeto:

FastAPI
React to FastAPICall a Python FastAPI from a React website
FastAPIAmazon DynamoDBPython
FastAPI to Python DynamoDBConnect a FastAPI to a DynamoDB table