Building a REST API with AWS Lambda URLs, Python, and AWS CDK

Yaroslav Murskyi
November 28, 2024

Introduction

AWS Lambda is a powerful serverless platform ideal for building small-scale REST services. There are three common methods to create a REST API with an AWS Lambda function: API Gateway, Application Load Balancer, and Lambda URLs (I’m not going to compare them here, but each has its pros and cons). This article will focus on the third option, Lambda URLs.

This guide walks you through building a user management service allowing clients to add, view, and delete users. We’ll utilize Python and the AWS CDK to develop, build and deploy the microservice. The microservice schema is as follows:

The microservice will have five endpoints for managing users, including Swagger documentation. We’ll store user data in memory for simplicity without relying on a database. Let’s start

Setting Up the AWS CDK Project

Let’s begin by constructing the microservice infrastructure. In my previous article , I explained the project structure and steps in creating a Lambda function using AWS CDK. I’ve considered that you already set up the local environment, so now we will dive into the details of how to configure the Lambda function URLs.

To define a Lambda function URL, we need to call the method add_function_url with the auth_type parameter. We will set it to None to provide public access to our Lambda function. The Lambda function also supports the AWS_IAM authentication type, which allows you to protect your function using AWS Identity and Access Management (IAM) to authenticate and authorize requests. The complete code for the constructor is provided below.

class DemoLambdaConstruct(Construct):
    def __init__(self, scope: Construct, construct_id: str) -> None:
        super().__init__(scope, construct_id)
        self.construct_id = construct_id
        self.lambda_function = self._build_lambda_function()
        self.lambda_function_url = self.lambda_function.add_function_url(
            auth_type=_lambda.FunctionUrlAuthType.NONE,
        )

    def _build_lambda_function(
        self,
    ) -> _lambda.Function:
        """
        Basic Python Lambda Function
        """
        return _lambda.Function(
            self,
            "BasicPythonLambdaFunction",
            function_name=self.construct_id,
            runtime=_lambda.Runtime.PYTHON_3_12,
            code=_lambda.Code.from_asset(constants.BUILD_FOLDER),
            handler="service.handlers.demo_lambda.lambda_handler",
            environment={
                "POWERTOOLS_SERVICE_NAME": "demo-service",
                "POWERTOOLS_TRACE_DISABLED": "true",
                "AWS_LAMBDA_LOG_LEVEL": constants.LOG_LEVEL,
            },
            tracing=_lambda.Tracing.DISABLED,
            retry_attempts=0,
            timeout=Duration.seconds(constants.HANDLER_LAMBDA_TIMEOUT),
            memory_size=constants.HANDLER_LAMBDA_MEMORY_SIZE,
            layers=[self._build_lambda_layer()],
            log_retention=RetentionDays.ONE_DAY,
            log_format=_lambda.LogFormat.JSON.value,
            system_log_level_v2=_lambda.SystemLogLevel.INFO,
            application_log_level_v2=_lambda.ApplicationLogLevel.DEBUG,
        )

    def _build_lambda_layer(self) -> PythonLayerVersion:
        """
        Build a Lambda Layer
        """
        return PythonLayerVersion(
            self,
            f"{self.construct_id}_layer",
            entry=constants.LAYER_BUILD_FOLDER,
            compatible_runtimes=[_lambda.Runtime.PYTHON_3_12],
            removal_policy=RemovalPolicy.DESTROY,
        )

Let’s proceed with creating API endpoints.

Defining API Endpoints

Our Lambda function will provide API endpoints for managing user entities. We will use the aws_lambda_powertools library to handle events, routing, and validation.

First, we need to define the URL resolver and enable Swagger documentation.

app = LambdaFunctionUrlResolver(enable_validation=True)
app.enable_swagger(path="/swagger", title="Demo Lambda", description="Demo Lambda API")

The Lambda handler will utilize the URL resolver that we initialized.

@logger.inject_lambda_context(correlation_id_path=correlation_paths.LAMBDA_FUNCTION_URL)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    """
    Lambda Handler
    """
    return app.resolve(event, context)

Let’s start with the endpoint that returns a list of users. This endpoint allows clients to filter the response by returning only active users or users with a specific role. The code for the API endpoint is provided below:

@app.get(
    "/users",
    summary="Get all users",
    description="API returns all users",
    response_description="All users",
    responses={
        HTTPStatus.OK: {
            "description": "All users",
            "content": {"application/json": {"model": List[User]}},
        }
    },
    tags=["users"],
)
def get_users(
    is_active: Optional[bool] = None,
    role: Annotated[Optional[UserRole], Query()] = None,
) -> List[User]:
    """
    Get all users with optional filtering by active status and role.

    Args:
        is_active (Optional[bool]): Filter by user active status
        role (Optional[UserRole]): Filter by user role

    Returns:
        List[User]: List of filtered users
    """
    filtered_users = [User(**user_data) for user_data in user_records]

    if is_active is not None:
        filtered_users = [user for user in filtered_users if user.active == is_active]

    if role is not None:
        filtered_users = [user for user in filtered_users if user.role == role]

    return filtered_users

The API endpoint retrieves all users from storage and filters the response based on any provided criteria. Let’s cover the endpoint with unit tests.

def test_get_users():
    response = lambda_handler(
        generate_api_lambda_event("/users", None), generate_context()
    )
    assert response["statusCode"] == HTTPStatus.OK
    response_body = json.loads(response["body"])
    assert len(response_body) == 10

    response = lambda_handler(
        generate_api_lambda_event(
            "/users", None, query_parameters={"is_active": "false"}
        ),
        generate_context(),
    )
    assert response["statusCode"] == HTTPStatus.OK
    response_body = json.loads(response["body"])
    assert len(response_body) == 2

The test checks the API endpoint and filter options. It uses the helper function generate_api_lambda_event, which generates a Lambda URL event and sends it to the Lambda handler. Now, let’s deploy the Lambda function and check the endpoint.

make deploy

Once you deploy the Lambda function URL, the deployment command will return a URL in the following format https://<url-id>.lambda-url.<region>.on.aws. You can now test the API endpoint using the provided URL.

http https://<url-id>.lambda-url.<region>.on.aws/users?role=customer&is_active=true

It returns all active customers.

[
    {
        "active": true,
        "email": "[email protected]",
        "role": "customer",
        "user_id": "550e8400-e29b-41d4-a716-446655440005"
    },
    {
        "active": true,
        "email": "[email protected]",
        "role": "customer",
        "user_id": "550e8400-e29b-41d4-a716-446655440007"
    },
    {
        "active": true,
        "email": "[email protected]",
        "role": "customer",
        "user_id": "550e8400-e29b-41d4-a716-446655440009"
    }
]

If you navigate to the URL https://<url-id>.lambda-url.<region>.on.aws/swagger in your browser, you will find the Swagger documentation for the endpoint.

Other endpoints are similar, except for the create and update endpoints, which include incoming request validation. You can find the full source code of the microservice on GitHub . To clone the project locally, use the following command:

git clone --branch lambda-url [email protected]:myarik/aws_cdk_python_demo.git
cd aws-cdk-python-demo

Resources

Powertools for AWS Lambda - Lambda Function URL : Powertools documentation.
Serverless API Documentation with Powertools for AWS : Describe how to set up Swagger documentation using PowerTools toolkits.

Conclusion

The aws_lambda_powertools toolkit and AWS CDK simplify the process of building REST microservices with AWS Lambda functions. The aws_lambda_powertools toolkit handles routing, validation, Swagger documentation, tracing, and more, enabling you to concentrate on developing your business logic. Similarly, AWS CDK allows you to easily build and manage your microservice infrastructure.

By leveraging these powerful tools, you can accelerate your development process, reduce operational overhead, and create scalable, maintainable applications.

How to Setup, Deploy, and Observe AWS Lambda Function in a Microservice Architecture

Yaroslav Murskyi
October 9, 2024

Introduction AWS Lambda is an excellent tool for building microservices due to its scalability, cost-efficiency, and maintainability. However, setting up, structuring, and monitoring a Lambda Function can be challenging.

Establishing a Secure Remote Development Environment with AWS EC2 and Terraform

Yaroslav Murskyi
November 17, 2023

Introduction Remote development environments (RDEs) allow software engineers to develop and deploy software remotely rather than on their local machine. RDEs are popular among software engineers for several reasons, including company security policies, requirements for specific resources, access to internal resources, and the ability to develop from different devices.

Building a Serverless Customer Support Ticket Routing Service

Yaroslav Murskyi
April 3, 2025

In this blog post, we will build a serverless customer support ticket routing service using AWS services like Lambda, API Gateway, SNS, and SQS.