OpenAPI Spec and AWS Lambda Powertools

Rasmus Fangel
3 min readJul 8, 2023

Supercharge your lambdas with OpenAPI Spec and Powertools!

Introduction

In this article, I will give a brief overview of what OpenAPI and AWS Lambda Powertools are, but if you are struggling with the basics of Python, OpenAPI or AWS Lambda Powertools, it is highly encouraged to take a look at the following articles and come back to this one:
- https://www.openapis.org/what-is-openapi
- https://github.com/aws-powertools/powertools-lambda-python

What is OpenAPI?

The OpenAPI Specification (formerly known as Swagger Specification) is a set of guidelines to standardise API contracts. It’s typically written in YAML or JSON and an example could look like this:

openapi: 3.0.3


info:
description: Info Description
version: "1.0.0"
title: Doc Title

paths:
/test/endpoint:
post:
summary: Summary
tags:
- Test
requestBody:
description: Description
required: true
content:
application/json:
schema:
type: object
required:
- param_1
properties:
param_1:
description: Param 1
type: string
responses:
"200":
description: Resp Description
content:
application/json:
schema:
type: object
properties:
message:
description: message_example
type: string

In the above example, we have defined an API contract saying that the request will look something like this:

HTTPMethod: POST,
Content-Type: "application/json",
Endpoint: "https://example.com/test/endpoint",
Body: { param_1: "example" } # required

We are also saying that if this endpoint is called, a JSON object with a message property will be sent back.

What is AWS Lambda Powertools?

Powertools provides you with a set of tools to supercharge your serverless lambdas. Take a look at the example below and notice how easy it is to set up tracing, logging (and more):

from typing import Optional

import requests
from requests import Response

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = APIGatewayRestResolver()


@app.get("/todos")
@tracer.capture_method
def get_todos():
todo_id: str = app.current_event.get_query_string_value(name="id", default_value="")
# alternatively
_: Optional[str] = app.current_event.query_string_parameters.get("id")

# Payload
_: Optional[str] = app.current_event.body # raw str | None

endpoint = "https://jsonplaceholder.typicode.com/todos"
if todo_id:
endpoint = f"{endpoint}/{todo_id}"

todos: Response = requests.get(endpoint)
todos.raise_for_status()

return {"todos": todos.json()}


# You can continue to use other utilities just as before
@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
return app.resolve(event, context)

The Problem

While Powertools is, for lack of better words, awesome, it does not allow you to verify your incoming lambda event against your OpenAPI contract. If you want to use the built-in validation that comes with Powertools, you will need to define a JSONSchema for the event to be validated against. This means, that you will have to maintain two seperate and very similar documents where the OpenAPI schema is “mostly for show”.

The Solution

As mentioned before, Powertools does not support, at present, validating your incoming lambda events against your OpenAPI Spec. To deal with that issue, I’ve written an alternative to the built-in validation that enables you to validate your OpenAPI Spec against your event.

powertools-oas-validator is a little tool that adds a decorator that wraps the lambda_handler and validates your OpenAPI Spec (OAS). Below is an example of how to use it:

from typing import Dict
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
from aws_lambda_powertools.utilities.typing import LambdaContext
from powertools_oas_validator.middleware import validate_request


app = APIGatewayRestResolver()

@app.post("/example")
def example() -> Response:
...

@validate_request(oas_path="openapi.yaml")
def lambda_handler(event: Dict, context: LambdaContext) -> Dict:
response = app.resolve(event, context)

return response

Please notice the use of the decorator @validate_request, where the first (and only) argument is the relative path to the OpenAPI Spec.

The tool parses the event and validates the event against it and will throw a SchemaValidationError if the validation fails. Below is an example of an error where the OAS defined a integer but a string was passed instead:

SchemaValidatonError(
name="test-path.test-endpoint.requestBody[param_1]",
path=["test-path", "test-endpoint", "requestBody", "param_1"],
validation_message="'not an integer' is not of type 'integer'.",
message="'not an integer' is not of type 'integer'",
rule="int",
rule_definition="type",
value="'not an integer'"
)

--

--

Rasmus Fangel

Software Engineer 👨‍💻 | Linux Enthusiast 🐧 | Gamer 🎮