Salta ai contenuti
@aws/nx-plugin
Blog

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 Gateway HTTP API di AWS. Configura AWS Lambda Powertools per l’osservabilità, inclusi logging, tracciamento AWS X-Ray e metriche Cloudwatch.

Utilizzo

Genera una FastAPI

Puoi generare una nuova FastAPI in due modi:

  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
    • Clicca su Generate

    Opzioni

    Parametro Tipo Predefinito Descrizione
    name Obbligatorio string - project name.
    directory string packages The directory to store the application in.

    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 progetto Python e dipendenze
    • Directory<module_name>
      • __init__.py Inizializzazione modulo
      • init.py Configura l’app FastAPI e il middleware powertools
      • main.py Implementazione API

    Il generatore crea anche costrutti CDK per il deployment dell’API, residenti nella directory packages/common/constructs.

    Implementare la tua FastAPI

    L’implementazione principale dell’API si trova in main.py. Qui si definiscono le route API e le relative implementazioni. 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:

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

    Osservabilità con AWS Lambda Powertools

    Logging

    Configurazione del logging strutturato tramite AWS Lambda Powertools. Accesso al logger negli 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
    • Percorso e metodo della richiesta
    • Informazioni sul contesto Lambda
    • Indicatori di cold start

    Tracing

    Tracciamento AWS X-Ray configurato automaticamente. Aggiunta di subsegmenti personalizzati:

    from .init import app, tracer
    

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

    Metriche

    Metriche CloudWatch raccolte automaticamente. Aggiunta 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}

    Metriche predefinite includono:

    • Conteggio richieste
    • Successi/fallimenti
    • Metriche cold start
    • Metriche per route specifiche

    Gestione Errori

    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}

    Eccezioni non gestite vengono:

    1. Registrate con stack trace
    2. Tracciate come metriche di fallimento
    3. Restituiscono risposta 500 sicura
    4. Mantengono l’ID di correlazione

    Streaming

    Con FastAPI puoi restituire risposte in streaming usando StreamingResponse.

    Modifiche Infrastrutturali

    Dato che API Gateway AWS non supporta lo streaming, dovrai deployare su piattaforme compatibili come AWS Lambda Function URL. Modificare il costrutto HttpApi generato:

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

    Dopo le modifiche, aggiornare packages/common/constructs/src/app/http-apis/<my-api>.ts per usare la nuova opzione function url.

    Implementazione

    Dopo l’aggiornamento infrastrutturale, implementa un API streaming in FastAPI:

    Esempio per lo streaming 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 stream di risposte, utilizza il Generatore API Connection per iterare in modo type-safe.

    Deploy della tua FastAPI

    Il generatore crea un costrutto CDK per il deployment in common/constructs. Utilizzo in un’applicazione CDK:

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

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

    Configura:

    1. Funzione Lambda con l’applicazione FastAPI
    2. API Gateway HTTP API come trigger
    3. Ruoli e permessi IAM
    4. Log group CloudWatch
    5. Configurazione tracciamento X-Ray
    6. Namespace metriche CloudWatch

    Concessione Accessi

    Metodo grantInvokeAccess per concedere accessi:

    api.grantInvokeAccess(myIdentityPool.authenticatedRole);

    Sviluppo Locale

    Server di sviluppo avviabile con:

    Terminal window
    pnpm nx run my-api:serve

    Include:

    • Auto-reload su modifiche
    • Documentazione interattiva in /docs o /redoc
    • Schema OpenAPI in /openapi.json

    Invocazione della tua FastAPI

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