How to build a RESTful API using Powertools for AWS Lambda (in Python)

Published: at 06:16 PM

Powertools for Lambda is a powerful developer toolkit with the aim of increase developer velocity and implement serverless best practices. In this project oriented article, we will build a personal finance RESTful API using Powertools for Lambda. I will walk you through some of the popular features of Powertools for Python. I will explain you also some of the under hood code of this development toolkit and what we can learn from it.

Table of contents

Open Table of contents

Preparation of the project

Powertools can be installed in different ways, but to get started in your project you can start using mainly the following ways: added as Lambda Layer, so after you deploy your project, your Lambda will have all the dependencies you need. By the other hand, as development dependency, so you can use them on development time but excluding it for production.

đź’ˇ Refer to this part of the article in future article to setup a basic CDK RESTful API python project. If the installation experience does not change I will link to this article on future project articles.

My first step will be create a CDK project, setup an AWS APIGateway connected with a Lambda (nothing fancy here). Let’s describe the steps to get there:

cdk init --language=python
# file: requirements-dev.txt

aws-lambda-powertools[all]
# Retrieve the reference of the Layer in the deployed region
# by ARN
powertools_layer = _lambda.LayerVersion.from_layer_version_arn(
    self,
    id="lambda-powertools",
    layer_version_arn=f"arn:aws:lambda:{self.region}:017000801446:layer:AWSLambdaPowertoolsPythonV2:32"
)
# Create lambda and add Power tools
transaction_handler_function = _lambda.Function(
    self,
    "TransactionsFunctionHandler",
    runtime=_lambda.Runtime.PYTHON_3_7,
    code=_lambda.Code.from_asset("functions/transactions"),
    handler="index.handler",
    layers=[powertools_layer],
)

And our lambda code initially will look like:

import aws_lambda_powertools
def handler(event, context):
    return {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": f"AWS Lambda powertools author: {aws_lambda_powertools.__author__}",
    }

This code just retrieves a successful response independently of the input event, adding in the body a sentence which contains the author of the development kit, which is a property that we can access importing the main Powertools library. After this, we can deploy our CDK project to our environment.

REST API

One of the most common Lambda integrations is with APIGateway for creating REST API’s. Usually, one of the infrastructure questions that you ask yourself when designing your REST API is about creating one single Lambda to handler all the resource paths or having one lambda per each endpoint. Both approaches have different trade-offs. If you want to read more about the main considerations to go for one approach or another, you have detail explanation here

For our finance app application, I start creating a REST API and add some endpoints to manage our transactions. A monolithic approach will suffice, which give us the opportunity to explore Powertools in depth.

If you coming from web development, you probably have seen before how web frameworks manage the incoming requests. Oversimplifying, the request hit the server, the main code of the applications boots up, the application process the request, if a middleware exist, it performs validations and transform the raw request on object class which pass to the application layer. Usually this is possible via resolver functions and a router registration, the object class transform from the request is redirected to a controller class and to a specific method which is able to read the request object class and process the business logic required to serve that request.

When we are working with Lambda we received in our handler method a raw event (or request event data in case of APIGW) and the context data from Lambda.

In monolithic Lambda, one of the problems we need to solve is how to run different logic depending on the endpoint that this events belongs, essentially how to implement a router mechanism, which as mentioned above in web frameworks this functionality is first citizen.

The first feature that we will take a look in Powertools is: Event Handlers. This feature solves among others the problems, the routing mechanism. Powertools is a very thin and fast layer of tools on top of your Lambda execution. I will show you some of the inner code that uses this feature. If you have some experience using web frameworks, the way this works will sound really familiar to you. Event Handlers offer us in overall, the following features:

The next step in our finance app is create the following endpoints in our APIGateway. All of them will hit the same Lambda code.

GET    /transactions?type=expense|income
POST   /transactions
GET    /transactions/{transaction_id}
DELETE /transactions/{transaction_id}

Each transaction resource will consist in the following resource model:

Transaction
  id: UUID
  amount: 500 (-500) # We will transform depending on the type
  type: expense | income
  category: str
  created_at: datetime

The next step is to modify our CDK project and to add all the resources that we described above:

# Create API Gateway - Lambda integration
api = apigw.LambdaRestApi(
    self,
    "FinanceAppApiLambdaIntegration",
    handler=transaction_handler_function,
)

# Set up endpoints
transactions = api.root.add_resource("transactions")
transactions.add_method("GET")  # GET /transactions
transactions.add_method("POST")  # POST /transactions

transaction = transactions.add_resource("{transaction_id}")
transaction.add_method("GET")  # GET /transactions/{transaction_id}
transaction.add_method("DELETE")  # DELETE /transactions/{transaction_id}

Next, we can start to use the APIGatewayRestResolver in our lambda. What we are going to do?:

from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.utilities.typing import LambdaContext

from transactions_controller import TransactionsController

app = APIGatewayRestResolver()
transactions_controller = TransactionsController()


@app.get("/transactions")
def list_transactions():
    return transactions_controller.list()


@app.post("/transactions")
def create_transaction():
    transaction_data: dict = app.current_event.json_body  # deserialize json str to dict
    return transactions_controller.create(
        transaction_data["amount"],
        transaction_data["type"],
        transaction_data["category"],
    )


@app.get("/transactions/<transaction_id>")
def get_transaction(transaction_id: str):
    return transactions_controller.find(transaction_id)


@app.delete("/transactions/<transaction_id>")
def delete_transaction(transaction_id: str):
    return transactions_controller.delete(transaction_id)


def handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)
lambda_arn = context.invoked_function_arn

And access easily to the variables of the lambda context.

Outside of our handler we instantiate APIGatewayRestResolver application which inherence to the ApiGatewayResolver and the BaseRouter. The ApiGatewayResolver do some interesting things. It registers the decorator function that we can use for our methods signatures using @app.{method} This register functions looks for all the items declared and for each one registered creates an instance of Route class. This route object is store in a collection of routes, so when the requests hit our lambda, the app.resolve will look for a route that matches the rule declared. This is possible because in the event received in the ApiGateway we have the path attribute which can be match against available route methods. Following you can see some of the code that allows that in the PowerTools base code:

# apigateway.py >> ApiGatewayResolver

for item in methods:
    _route = Route(item, self._compile_regex(rule), func, cors_enabled, compress, cache_control)

    # The more specific route wins.
    # We store dynamic (/studies/{studyid}) and static routes (/studies/fetch) separately.
    # Then attempt a match for static routes before dynamic routes.
    # This ensures that the most specific route is prioritized and processed first (studies/fetch).
    if _route.rule.groups > 0:
        self._dynamic_routes.append(_route)
    else:
        self._static_routes.append(_route)

# apigateway.py >> ApiGatewayResolver

def _resolve(self) -> ResponseBuilder:
....
   for route in self._static_routes + self._dynamic_routes:
       if method != route.method:
        continue
    match_results: Optional[Match] = route.rule.match(path)
    if match_results:
        logger.debug("Found a registered route. Calling function")

Another feature which is included as part of the routing is have the ability to also take in account path parameters, like for instance in the case of @app.get("/transactions/<transaction_id>") which <transaction_id> will be recognise for the routing mechanism as dynamic variable and it will try to inject on the function when calling the method. So for instance, we can do:

@app.get("/transactions/&lt;transaction_id&gt;")
def get_transaction(transaction_id: str):
   ....

And use directly the transaction_id value without parsing the entry request as we would do normally like:

def handler(event, context):
    transaction_id = event["pathParameters"]["transaction_id"]

What happen for the POST methods where is going to be a payload on the event? As we mentioned before PowerTools wraps the context in a class, but on resolve time can do also the same with the event. The event will be place on app.current_event and which will be a class object from APIGatewayProxyEvent and one of the class properties of this class is json_body which basically performs a deserialization of the event[“body”] and converts it in a Python dict. So the only code we need to write is:

transaction_data: dict = app.current_event.json_body  # deserialize json str to dict

What about the responses? We did not write any code to create a response for the Lambda. This is because Powertools resolve the response for us using default serializers. Let’s look to some of the internal code:

# After the handler call app.resolve(event, context) method

def resolve(self, event, context) -&gt; Dict[str, Any]:
    ... some code
    response = self._resolve().build(self.current_event, self._cors)

The _resolve() method call the route based on the registered routes, the route method is executed and the result is wrap on a Respond object. This Response object has a method called build which if we look at it:

.... other code inside the build method

return {
    "statusCode": self.response.status_code,
    "body": self.response.body,
    "isBase64Encoded": self.response.base64_encoded,           **event.header_serializer().serialize(headers=self.response.headers, cookies=self.response.cookies),
}

Among other things, prepare the response as we had to do manually, encode the response and add the required headers. Unless we need to write custom responses we do not have to write our own responses. Like a mentioned before the way this is built probably will sound familiar to you, if you work before with web frameworks.

Tracing

Another capability of Powertools is add tracing via AWS X-Ray. To do that we need to add the permissions to our lambda definition to active traces with lambda.Tracing.ACTIVE and also add a lambda environment variable which with we will identify the trace create by our lambda. By default, Powertools expect the environment variable POWERTOOL_SERVICE_NAME but it can be override. Adding the environment variable we won’t need to do any more setup.

from aws_lambda_powertools import Tracer
…
tracer = Tracer()
…
@app.get("/transactions")
@tracer.capture_method
def list_transactions():
return transactions_controller.list()

capture_method will capture and send the telemetry information to X-Ray when the method is hit. How will identify in X-Ray this specific method information? If we open the main code of the capture_method and read the comments you will see that comments mention that capture_method will creates a subsegment named ## <method_module.method_qualifiedname> which uses a Python implementation to qualified name for classes and functions that can be found here: https://peps.python.org/pep-3155/ which is a proposal that is based on create a qualifier attribute called __qualname__ for functions and classes which helps to get a dotted path leading to the object from the module top-level. For instance, in this example extracted from the link:

&gt;&gt;&gt; class C:
...   def f(): pass
...   class D:
...     def g(): pass

&gt;&gt;&gt; C.D.g.__qualname__
'C.D.g'

Now if we go to our http client and hit our endpoint for /transactions we will see the tracing information in the X-Ray console.

We can see the Powertools create a subsegment for the top-level function and also for the called function that underlines it using the qualifiers. If you want to add annotations or metadata, you can use the method put_metadata or put_annotations of the Trace of object for this. Remember that annotations are indexed, so you can group traces based on this and metadata is not indexed. This example add to the metadata the transaction_id look on the call:

@app.get("/transactions/&lt;transaction_id&gt;")
@tracer.capture_method
def get_transaction(transaction_id: str):
    tracer.put_annotation(key="transaction_id", value=transaction_id)

    return transactions_controller.find(transaction_id)

Now if we hit the endpoint and we back to the X-Ray we can search for that particular transaction_id and find the trace that we just created.

Exceptions

Until now we cover how Powertools help us with the routing and with the traceability of our code. Another important aspect in any application but specially on REST apis is handling exceptions. Let’s see how PowerTools helps us again on that matter, so we do not have to write

# index.py

@app.exception_handler(ItemNotFoundException)
def handle_item_notfound_exception(ex: ItemNotFoundException):
    raise NotFoundError(msg=str(ex))

Another software design benefit is that we can make the dependencies go from inside to outside. So our business logic does not know anything about Powertools or its exceptions. Other built exceptions Powertools offer us are: BadRequestError, InternalServerError, NotFoundError, ServiceError, UnauthorizedError.

Logging

Logging is the last part of our journey. One of the keys of a great development experience is to have instrumentation which can help us to keep our code clean and less complex. In case of Lambda development, it’s even more important have this intrumentation, because the Lambda context contains information per request that we can use to detect issues and benchmark different implementations. As other features in Powertools logging starts with instantiating the Logger class. The main features that we can use in our code for logging start with include in our logs the context information. To do that, we can use a decorator in our handler function which will include all the context data in our log:

@logger.inject_lambda_context
@tracer.capture_lambda_handler
def handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)

Another interesting feature on development time is to log the input event in every resquest. This is possible, enabling a flag on the inject_lambda_context This feature is not recommended on production. This can cause a high amount of logging besides from the risk of logging sensitive information.

@logger.inject_lambda_context(log_event=True)
@tracer.capture_lambda_handler
def handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)

The interface offer us different levels of logging: info, error, exception, critical, warning, debug. Which depending on the active logging severity we will have those logs in our CloudWatch.

When we add new custom keys to our logs we need to take in account that they are consistent across the warm invocations, so we need to make sure that the values of the keys reflect the latest invocation values. Powertools give us different ways to append keys:

Method 1:

# Persist across invocation.

logger.append_keys(transaction_id=transaction_id)

Method 2 (two ways):

# Per log basis

logger.info("Fetching transaction", transaction_id=transaction_id)

# Using extra method

logger.info("Fetching transaction", extra={"transaction_id": transaction_id})

The main different between these two methods is that append_keys will be persistent keys across different warm invocations, so I think it can be very useful for common logging values that are going to be present in every case invocation and are potentially appear in the handler method itself or arguments that are passed on our event handler methods. Persisting information across invocations can increase the size of our loggings, so in my opinion we need to think where to use it. One approach is to locate minimal persistent information which will be added to the ephemeral information enriching the logging information.