OpenAPI Spec and AWS Lambda Powertools
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'"
)