FastAPI

FastAPI è un framework per la creazione di API in Python.

Il generatore FastAPI crea una nuova applicazione FastAPI con configurazione infrastrutturale AWS CDK. Il backend generato utilizza AWS Lambda per il deployment serverless, esposto tramite un’API AWS API Gateway. Configura AWS Lambda Powertools per l’osservabilità, inclusi logging, tracciamento AWS X-Ray e metriche Cloudwatch.

Utilizzo

Generare una FastAPI

Puoi generare una nuova FastAPI in due modi:

VSCode

1. Installa il Nx Console VSCode Plugin se non l'hai già fatto
2. Apri la console Nx in VSCode
3. Clicca su Generate (UI) nella sezione "Common Nx Commands"
4. Cerca @aws/nx-plugin - py#fast-api
5. Compila i parametri richiesti
6. Clicca su 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

Puoi anche eseguire una prova per vedere quali file verrebbero modificati

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

Opzioni

Parametro	Tipo	Predefinito	Descrizione
name Obbligatorio	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.

Nota

Puoi scegliere tra una REST API o una HTTP API per distribuire la tua API. Fai riferimento alla API Gateway Developer Guide per aiutarti a decidere.

Output del Generatore

Il generatore creerà la seguente struttura del progetto nella directory <directory>/<api-name>:

• project.json Configurazione del progetto e target di build
• pyproject.toml Configurazione del progetto Python e dipendenze
• Directory<module_name>

  • __init__.py Inizializzazione del modulo
  • init.py Configura l’app FastAPI e il middleware powertools
  • main.py Implementazione dell’API
• Directoryscripts

  • generate_open_api.py Script per generare uno schema OpenAPI dall’app FastAPI

Il generatore creerà anche costrutti CDK utilizzabili per deployare l’API, residenti nella directory packages/common/constructs.

Implementazione della tua FastAPI

L’implementazione principale dell’API si trova in main.py. Qui definisci le route API e le relative implementazioni. Ecco un esempio:

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

Il generatore configura automaticamente diverse funzionalità:

1. Integrazione AWS Lambda Powertools per l’osservabilità
2. Middleware per la gestione degli errori
3. Correlazione richiesta/risposta
4. Raccolta metriche
5. Handler AWS Lambda utilizzando Mangum

Osservabilità con AWS Lambda Powertools

Logging

Il generatore configura il logging strutturato utilizzando AWS Lambda Powertools. Puoi accedere al logger nei tuoi route handler:

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}

Il logger include automaticamente:

• ID di correlazione per il tracciamento delle richieste
• Percorso e metodo della richiesta
• Informazioni sul contesto Lambda
• Indicatori di cold start

Tracciamento

Il tracciamento AWS X-Ray è configurato automaticamente. Puoi aggiungere sottosegmenti personalizzati:

from .init import app, tracer

@app.get("/items/{item_id}")
@tracer.capture_method
def read_item(item_id: int):
    # Crea un nuovo sottosegmento
    with tracer.provider.in_subsegment("fetch-item-details"):
        # Logica qui
        return {"item_id": item_id}

Metriche

Le metriche CloudWatch vengono raccolte automaticamente per ogni richiesta. Puoi aggiungere metriche personalizzate:

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}

Le metriche predefinite includono:

• Conteggi richieste
• Conteggi successo/fallimento
• Metriche cold start
• Metriche per route specifiche

Gestione degli Errori

Il generatore include una gestione errori completa:

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}

Le eccezioni non gestite vengono intercettate dal middleware e:

1. Registrano l’eccezione completa con stack trace
2. Registrano una metrica di fallimento
3. Restituiscono una risposta sicura 500 al client
4. Preservano l’ID di correlazione

Consiglio

Si raccomanda di specificare modelli di risposta per le operazioni API per una migliore generazione del codice se si utilizza il generatore api-connection. Vedi qui per maggiori dettagli.

Streaming

Con FastAPI, puoi inviare una risposta in streaming al chiamante utilizzando il tipo di risposta StreamingResponse.

Modifiche Infrastrutturali

Poiché AWS API Gateway non supporta risposte in streaming, dovrai deployare la tua FastAPI su una piattaforma che lo supporti. L’opzione più semplice è utilizzare un AWS Lambda Function URL. Per farlo, puoi sostituire il costrutto generato common/constructs/src/app/apis/<name>-api.ts con uno che deploya un Function URL.

Esempio di Costrutto FunctionURL per 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 });

    // Registra l'URL dell'API nella configurazione di runtime per la scoperta del client
    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',
        },
      },
    });
  }
}

Nota

Per un esempio completo, consulta il Tutorial Dungeon Adventure

Implementazione

Dopo aver aggiornato l’infrastruttura per supportare lo streaming, puoi implementare un’API in streaming con FastAPI. L’API dovrebbe:

• Restituire un StreamingResponse
• Dichiarare il tipo di ritorno per ogni chunk della risposta
• Aggiungere l’estensione vendor OpenAPI x-streaming: true se intendi utilizzare il API Connection.

Ad esempio, per inviare in streaming una serie di oggetti 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

Per consumare uno stream di risposte, puoi utilizzare il Generatore API Connection che fornirà un metodo type-safe per iterare sui chunk in streaming.

Deploy della tua FastAPI

Il generatore FastAPI crea un costrutto CDK per il deploy nella cartella common/constructs. Puoi usarlo in un’applicazione CDK:

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

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // Aggiungi l'API allo stack
    const api = new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this).build(),
    });
  }
}

Questo configura:

1. Una funzione AWS Lambda per ogni operazione nell’app FastAPI
2. API Gateway HTTP/REST API come trigger della funzione
3. Ruoli e permessi IAM
4. Log group CloudWatch
5. Configurazione tracciamento X-Ray
6. Namespace per metriche CloudWatch

Nota

Se hai selezionato l’autenticazione Cognito, dovrai fornire la proprietà identity al costrutto API:

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

Il costrutto UserIdentity può essere generato usando il generatore ts#cloudscape-website-auth

Integrazioni Type-Safe

I costrutti CDK per le API REST/HTTP sono configurati per fornire un’interfaccia type-safe per definire le integrazioni per ciascuna delle tue operazioni.

Integrazioni predefinite

Puoi utilizzare il metodo statico defaultIntegrations per sfruttare il pattern predefinito, che definisce una singola funzione AWS Lambda per ogni operazione:

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

Accesso alle integrazioni

Puoi accedere alle funzioni AWS Lambda sottostanti tramite la proprietà integrations del costrutto API, in modo type-safe. Ad esempio, se la tua API definisce un’operazione chiamata sayHello e hai bisogno di aggiungere alcune autorizzazioni a questa funzione, puoi farlo come segue:

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

// sayHello è tipizzato in base alle operazioni definite nella tua API
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: [...],
  resources: [...],
}));

Personalizzazione delle opzioni predefinite

Se desideri personalizzare le opzioni utilizzate durante la creazione della funzione Lambda per ogni integrazione predefinita, puoi usare il metodo withDefaultOptions. Ad esempio, se vuoi che tutte le tue funzioni Lambda risiedano in una Vpc:

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

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

Override delle integrazioni

Puoi anche sovrascrivere le integrazioni per operazioni specifiche usando il metodo withOverrides. Ogni override deve specificare una proprietà integration tipizzata secondo l’appropriato costrutto CDK di integrazione per l’API HTTP o REST. Il metodo withOverrides è anch’esso type-safe. Ad esempio, se desideri sovrascrivere un’API getDocumentation per puntare a documentazione ospitata su un sito esterno:

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

Noterai che l’integrazione sovrascritta non avrà più una proprietà handler quando vi accedi tramite api.integrations.getDocumentation.

Puoi aggiungere proprietà aggiuntive a un’integrazione che saranno anch’esse tipizzate correttamente, permettendo l’astrazione di altri tipi di integrazione mantenendo il type-safety. Ad esempio, se hai creato un’integrazione S3 per un’API REST e successivamente vuoi riferirti al bucket per una specifica operazione:

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

// Successivamente, magari in un altro file, puoi accedere alla proprietà bucket che abbiamo definito
// in modo type-safe
api.integrations.getFile.bucket.grantRead(...);

Override degli autorizzatori

Puoi anche fornire options nelle tue integrazioni per sovrascrivere opzioni specifiche dei metodi come gli autorizzatori. Ad esempio, se desideri usare l’autenticazione Cognito per l’operazione getDocumentation:

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

Integrazioni esplicite

Se preferisci, puoi scegliere di non usare le integrazioni predefinite e fornirne direttamente una per ogni operazione. Questo è utile se, ad esempio, ogni operazione necessita di un tipo diverso di integrazione o vuoi ricevere un errore di tipo quando aggiungi nuove operazioni:

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

Pattern Router

Se preferisci distribuire una singola funzione Lambda per gestire tutte le richieste API, puoi modificare liberamente il metodo defaultIntegrations della tua API per creare una singola funzione invece di una per integrazione:

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 {
          // Riferimento allo stesso router lambda handler in ogni integrazione
          integration: new LambdaIntegration(router),
        };
      },
    });
  };
}

Puoi modificare il codice in altri modi se preferisci, ad esempio potresti definire la funzione router come parametro di defaultIntegrations invece di costruirla all’interno del metodo.

Generazione Codice

Poiché le operazioni in FastAPI sono definite in Python e l’infrastruttura in TypeScript, utilizziamo la generazione di codice per fornire metadati al costrutto CDK e creare un’interfaccia type-safe per le integrazioni.

Un target generate:<ApiName>-metadata viene aggiunto al project.json dei costrutti comuni per facilitare questa generazione, producendo un file come packages/common/constructs/src/generated/my-api/metadata.gen.ts. Essendo generato al build time, viene ignorato dal version control.

Nota

Dovrai eseguire un build dopo ogni modifica all’API per garantire l’aggiornamento dei tipi consumati dal costrutto CDK.

pnpm

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

yarn

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

npm

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

bun

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

Consiglio

Se stai lavorando contemporaneamente all’infrastruttura CDK e alla FastAPI, puoi usare nx watch per rigenerare i tipi ad ogni modifica:

pnpm

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

yarn

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

npm

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

bun

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

Concessione Accesso (Solo IAM)

Se hai selezionato l’autenticazione IAM, puoi usare il metodo grantInvokeAccess per concedere l’accesso:

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

Sviluppo Locale

Il generatore configura un server di sviluppo locale eseguibile con:

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

Questo avvia un server di sviluppo FastAPI con:

• Auto-reload alle modifiche al codice
• Documentazione API interattiva su /docs o /redoc
• Schema OpenAPI su /openapi.json

Invocazione della tua FastAPI

Per invocare l’API da un sito React, puoi usare il generatore api-connection.