Skip to content

Python Lambda Functions

The Python Lambda Function generator provides the ability to add a lambda function to an existing python project.

This generator creates a new python lambda handler with AWS CDK infrastructure setup. The generated backend uses AWS Lambda for serverless deployment, with optional type-safety using the Parser from AWS Lambda Powertools. It sets up AWS Lambda Powertools for observability, including logging, AWS X-Ray tracing and Cloudwatch Metrics.

Usage

Generate a Lambda Function

You can generate a new Lambda Function in two ways:

  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#lambda-function
  5. Fill in the required parameters
    • Click Generate

    Options

    Parameter Type Default Description
    project required string - The project to add the lambda function to
    functionName required string - The name of the function to add
    functionPath string - The path within the project source directory to add the function to
    eventSource required string Any Optional event source model to use for the lambda function

    Generator Output

    The generator will add the following files to your project:

    • Directory<module-name>
      • <lambda-function>.py Function implementation

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

    If the functionPath option is provided, the generater will add the necessary files to the specified path:

    • Directory<module-name>
      • Directory<custom-path>
        • <function-name>.py Function implementation

    Implementing your Function

    The main function implementation is in <function-name>.py. Here’s an example:

    import os
    from aws_lambda_powertools import Logger, Metrics, Tracer
    from aws_lambda_powertools.metrics import MetricUnit
    from aws_lambda_powertools.utilities.parser import event_parser
    from aws_lambda_powertools.utilities.parser.models import EventBridgeModel
    from aws_lambda_powertools.utilities.typing import LambdaContext
    os.environ["POWERTOOLS_METRICS_NAMESPACE"] = "Foo"
    os.environ["POWERTOOLS_SERVICE_NAME"] = "Foo"
    logger: Logger = Logger()
    metrics: Metrics = Metrics()
    tracer: Tracer = Tracer()
    @tracer.capture_lambda_handler
    @metrics.log_metrics
    @event_parser(model=EventBridgeModel)
    def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    logger.info("Received event", extra={"event": event.model_dump() })
    metrics.add_metric(name="InvocationCount", unit=MetricUnit.Count, value=1)
    try:
    # TODO: Implement
    metrics.add_metric(name="SuccessCount", unit=MetricUnit.Count, value=1)
    # TODO: Implement success response if required
    except Exception as e:
    logger.exception(e)
    metrics.add_metric(name="ErrorCount", unit=MetricUnit.Count, value=1)
    # TODO: Implement error response if required

    The generator sets up several features automatically:

    1. AWS Lambda Powertools integration for observability
    2. Metrics collection
    3. Type-safety using @event_parser

    Observability with AWS Lambda Powertools

    Logging

    The generator configures structured logging using AWS Lambda Powertools.

    def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    logger.info("Received event", extra={"event": event.model_dump()})

    The logger automatically includes:

    • Event requests
    • Lambda context information
    • Cold start indicators

    Tracing

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

    def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # Creates a new subsegment
    with tracer.provider.in_subsegment("function-subsegment"):
    # Your logic here
    return ....

    Metrics

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

    def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    metrics.add_metric(name="NewMetric", unit=MetricUnit.Count, value=1)
    return ...

    Default metrics include:

    • Invocation counts
    • Success/failure counts
    • Cold start metrics

    Type Safety

    If you chose an eventSource when generating your lambda function, your function is instrumented with @event_parser from AWS Lambda Powertools. For example:

    @event_parser(model=EventBridgeModel)
    def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    event.detail_type # <- type-safe with IDE autocompletion

    This allows you to define data models using Pydantic, in a similar manner to working with FastAPI.

    If you do not wish to type your event, you can just select Any for your eventSource.

    Deploying your Function

    The Python Lambda Function generator creates a CDK construct for deploying your function in the common/constructs folder. You can use this in a CDK application:

    import { MyProjectMyFunction } from ':my-scope/common-constructs';
    export class ExampleStack extends Stack {
    constructor(scope: Construct, id: string) {
    // Add the function to your stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');
    }
    }

    This sets up:

    1. AWS Lambda function
    2. CloudWatch log group
    3. X-Ray tracing configuration
    4. CloudWatch metrics namespace

    This function can then be used as a target for any lambda event source:

    The example below demonstrates the CDK code for invoking a your lambda function on a schedule using Event Bridge:

    import { EventPattern, Rule, Schedule } from 'aws-cdk-lib/aws-events';
    import { LambdaFunction } from 'aws-cdk-lib/aws-events-targets';
    import { MyProjectMyFunction } from ':my-scope/common-constructs';
    export class ExampleStack extends Stack {
    constructor(scope: Construct, id: string) {
    // Add the function to your stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');
    // Add the function to an EventBridge task
    const eventRule = new Rule(this, 'MyFunctionScheduleRule', {
    schedule: Schedule.cron({ minute: '15' }),
    targets: [new LambdaFunction(fn)],
    });
    }
    }