FastAPI

FastAPI is a framework for building APIs in Python.

The FastAPI generator creates a new FastAPI with AWS CDK infrastructure setup. The generated backend uses AWS Lambda for serverless deployment, exposed via an AWS API Gateway API. It sets up AWS Lambda Powertools for observability, including logging, AWS X-Ray tracing and Cloudwatch Metrics.

Usage

Generate a FastAPI

You can generate a new FastAPI in two ways:

VSCode

1. Install the Nx Console VSCode Plugin if you haven't already
2. Open the Nx Console in VSCode
3. Click Generate (UI) in the "Common Nx Commands" section
4. Search for @aws/nx-plugin - py#fast-api
5. Fill in the required parameters
6. Click 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

You can also perform a dry-run to see what files would be changed

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

Options

Parameter	Type	Default	Description
name Required	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.

Note

You can choose between either a REST API or HTTP API to deploy your API, please refer to the API Gateway Developer Guide to help you decide.

Generator Output

The generator will create the following project structure in the <directory>/<api-name> directory:

• project.json Project configuration and build targets
• pyproject.toml Python project configuration and dependencies
• Directory<module_name>

  • __init__.py Module initialisation
  • init.py Sets the up FastAPI app and configures powertools middleware
  • main.py API implementation
• Directoryscripts

  • generate_open_api.py Script to generate an OpenAPI schema from the FastAPI app

The generator will also create CDK constructs which can be used to deploy your API, which reside in the packages/common/constructs directory.

Implementing your FastAPI

The main API implementation is in main.py. This is where you define your API routes and their implementations. Here’s an example:

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

The generator sets up several features automatically:

1. AWS Lambda Powertools integration for observability
2. Error handling middleware
3. Request/response correlation
4. Metrics collection
5. AWS Lambda handler using Mangum

Observability with AWS Lambda Powertools

Logging

The generator configures structured logging using AWS Lambda Powertools. You can access the logger in your route 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}

The logger automatically includes:

• Correlation IDs for request tracing
• Request path and method
• Lambda context information
• Cold start indicators

Tracing

AWS X-Ray tracing is configured automatically. You can add custom subsegments to your traces:

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}

Metrics

CloudWatch metrics are collected automatically for each request. You can add custom metrics:

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}

Default metrics include:

• Request counts
• Success/failure counts
• Cold start metrics
• Per-route metrics

Error Handling

The generator includes comprehensive error handling:

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}

Unhandled exceptions are caught by the middleware and:

1. Log the full exception with stack trace
2. Record a failure metric
3. Return a safe 500 response to the client
4. Preserve the correlation ID

Tip

It’s recommended to specify response models for your API operations for better code generation if using the api-connection generator. See here for more details.

Streaming

With FastAPI, you can stream a response to the caller with the StreamingResponse response type.

Infrastructure Changes

Since AWS API Gateway does not support streaming responses, you will need to deploy your FastAPI to a platform which supports this. The simplest option is to use an AWS Lambda Function URL. To achieve this, you can replace the generated common/constructs/src/app/apis/<name>-api.ts construct with one that deploys a function URL instead.

Example Streaming FunctionURL Construct

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

    // Register the API URL in runtime configuration for client discovery
    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',
        },
      },
    });
  }
}

Note

For an end-to-end example, please refer to the Dungeon Adventure Tutorial

Implementation

Once you’ve updated the infrastructure to support streaming, you can implement a streaming API in FastAPI. The API should:

• Return a StreamingResponse
• Declare the return type of each response chunk
• Add the OpenAPI vendor extension x-streaming: true if you intend to use the API Connection.

For example, if you would like to stream a series of JSON objects from your API, you can implement this as follows:

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")

Consumption

To consume a stream of responses, you can make use of the API Connection Generator which will provide a type-safe method for iterating over your streamed chunks.

Deploying your FastAPI

The FastAPI generator creates a CDK construct for deploying your API in the common/constructs folder. You can use this in a CDK application:

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

This sets up:

1. An AWS Lambda function for each operation in the FastAPI application
2. API Gateway HTTP/REST API as the function trigger
3. IAM roles and permissions
4. CloudWatch log group
5. X-Ray tracing configuration
6. CloudWatch metrics namespace

Note

If you selected to use Cognito authentication, you will need to supply the identity property to the API construct:

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

The UserIdentity construct can be generated using the ts#cloudscape-website-auth generator

Type-Safe Integrations

The REST/HTTP API CDK constructs are configured to provide a type-safe interface for defining integrations for each of your operations.

Default Integrations

You can use the static defaultIntegrations to make use of the default pattern, which defines an individual AWS Lambda function for each operation:

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

Accessing Integrations

You can access the underlying AWS Lambda functions via the API construct’s integrations property, in a type-safe manner. For example, if your API defines an operation named sayHello and you need to add some permissions to this function, you can do so as follows:

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: [...],
}));

Customising Default Options

If you would like to customise the options used when creating the Lambda function for each default integration, you can use the withDefaultOptions method. For example, if you would like all of your Lambda functions to reside in a Vpc:

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

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

Overriding Integrations

You can also override integrations for specific operations using the withOverrides method. Each override must specify an integration property which is typed to the appropriate CDK integration construct for the HTTP or REST API. The withOverrides method is also type-safe. For example, if you would like to override a getDocumentation API to point to documentation hosted by some external website you could achieve this as follows:

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

You will also notice that the overridden integration no longer has a handler property when accessing it via api.integrations.getDocumentation.

You can add additional properties to an integration which will also be typed accordingly, allowing for other types of integration to be abstracted but remain type-safe, for example if you have created an S3 integration for a REST API and later wish to reference the bucket for a particular operation, you can do so as follows:

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(...);

Overriding Authorizers

You can also supply options in your integration to override particular method options such as authorizers, for example if you wished to use Cognito authentication for your getDocumentation operation:

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

Explicit Integrations

If you prefer, you can choose not to use the default integrations and instead directly supply one for each operation. This is useful if, for example, each operation needs to use a different type of integration or you would like to receive a type error when adding new operations:

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

Router Pattern

If you prefer to deploy a single Lambda function to service all API requests, you can freely edit the defaultIntegrations method for your API to create a single function instead of one per integration:

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 {
          // Reference the same router lambda handler in every integration
          integration: new LambdaIntegration(router),
        };
      },
    });
  };
}

You can modify the code in other ways if you prefer, for example you may prefer to define the router function as a parameter to defaultIntegrations instead of constructing it within the method.

Code Generation

Since operations in FastAPI are defined in Python and infrastructure in TypeScript, we instrument code-generation to supply metadata to the CDK construct to provide a type-safe interface for integrations.

A generate:<ApiName>-metadata target is added to the common constructs project.json to facilitate this code generation, which emits a file such as packages/common/constructs/src/generated/my-api/metadata.gen.ts. Since this is generated at build time, it is ignored in version control.

Note

You will need to run a build whenever you change your API to ensure the types consumed by the CDK construct are up to date.

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

Tip

If you are actively working on both your CDK infrastructure and FastAPI together, you can use nx watch to regenerate these types every time you make API changes:

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"

Granting Access (IAM Only)

If you selected to use IAM authentication, you can use the grantInvokeAccess method to grant access to your API:

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

Local Development

The generator configures a local development server that you can run with:

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

This starts a local FastAPI development server with:

• Auto-reload on code changes
• Interactive API documentation at /docs or /redoc
• OpenAPI schema at /openapi.json

Invoking your FastAPI

To invoke your API from a React website, you can use the api-connection generator.