đλ⚠- lambda_decorators¶
A collection of useful decorators for making AWS Lambda handlers
lambda_decorators
is a collection of useful decorators for writing Python
handlers for AWS Lambda. They allow you to
avoid boiler plate for common things such as CORS headers, JSON serialization,
etc.
Quick example¶
# handler.py
from lambda_decorators import json_http_resp, load_json_body
@json_http_resp
@load_json_body
def handler(event, context):
return {'hello': event['body']['name']}
When deployed to Lambda behind API Gateway and cURLâd:
$ curl -d '{"name": "world"}' https://example.execute-api.us-east-1.amazonaws.com/dev/hello
{"hello": "world"}
Install¶
If you are using the serverless framework I recommend using serverless-python-requirements
sls plugin install -n serverless-python-requirements
echo lambda-decorators >> requirements.txt
Or if using some other deployment method to AWS Lambda you can just download the entire module because itâs only one file.
curl -O https://raw.githubusercontent.com/dschep/lambda-decorators/master/lambda_decorators.py
Included Decorators:¶
lambda_decorators
includes the following decorators to avoid boilerplate
for common usecases when using AWS Lambda with Python.
async_handler()
- support for async handlerscors_headers()
- automatic injection of CORS headersdump_json_body()
- auto-serialization of http body to JSONload_json_body()
- auto-deserialize of http body from JSONjson_http_resp()
- automatic serialization of python object to HTTP JSON responsejson_schema_validator()
- use JSONSchema to validate request&response payloadsload_urlencoded_body()
- auto-deserialize of http body from a querystring encoded bodyno_retry_on_failure()
- detect and stop retry attempts for scheduled lambdasssm_parameter_store()
- fetch parameters from the AWS SSM Parameter Storesecrets_manager()
- fetch secrets from the AWS Secrets Manager
See each individual decorators for specific usage details and the example for some more use cases. This library is also meant to serve as an example for how to write decorators for use as lambda middleware. See the recipes page for some more niche examples of using decorators as middleware for lambda.
Writing your own¶
lambda_decorators
includes utilities to make building your own decorators
easier. The before()
, after()
, and on_exception()
decorators
can be applied to your own functions to turn them into decorators for your
handlers. For example:
import logging
from lambda_decorators import before
@before
def log_event(event, context):
logging.debug(event)
return event, context
@log_event
def handler(event, context):
return {}
And if you want to make a decorator that provides two or more of
before/after/on_exception functionality, you can use
LambdaDecorator
:
import logging
from lambda_decorators import LambdaDecorator
class log_everything(LambdaDecorator):
def before(event, context):
logging.debug(event, context)
return event, context
def after(retval):
logging.debug(retval)
return retval
def on_exception(exception):
logging.debug(exception)
return {'statusCode': 500}
@log_everything
def handler(event, context):
return {}
Why¶
Initially, I was inspired by middy which I like using in JavaScript. So naturally, I thought Iâd like to have something similar in Python too. But then as I thought about it more, it seemed that when thinking of functions as the compute unit, when using python, decorators pretty much are middleware! So instead of building a middleware engine and a few middlewares, I just built a few useful decorators and utilities to build them.
-
class
lambda_decorators.
LambdaDecorator
(handler)[source]¶ This is a class for simplifying the creation of decorators for use on Lambda handlers.
You subclass
LambdaDecorator
and add your ownbefore()
,after()
, oron_exception()
methods.Usage:
>>> from lambda_decorators import LambdaDecorator >>> class print_event_and_resp(LambdaDecorator): ... def before(self, event, context): ... print('event: ', event) ... return event, context ... def after(self, retval): ... print('retval: ', retval) ... return retval >>> @print_event_and_resp ... def handler(event, context): ... return 'hello world' >>> handler({'foo': 'bar'}, object()) event: {'foo': 'bar'} retval: hello world 'hello world' >>> >>> # And exception handling: >>> class handle_exceptions(LambdaDecorator): ... def on_exception(self, exception): ... return {'statusCode': 500, 'body': 'uh oh, you broke it'} >>> @handle_exceptions ... def handler(event, context): ... raise Exception >>> handler({}, object) {'statusCode': 500, 'body': 'uh oh, you broke it'}
-
lambda_decorators.
after
(func)[source]¶ Run a function after the handler is invoked, is passed the response and must return an response too.
Usage:
>>> # to create a reusable decorator >>> @after ... def gnu_terry_pratchett(retval): ... retval.setdefault('Headers', {})['X-Clacks-Overhead'] = 'GNU Terry Pratchett' ... return retval >>> @gnu_terry_pratchett ... def handler(event, context): ... return {'body': ''} >>> handler({}, object()) {'body': '', 'Headers': {'X-Clacks-Overhead': 'GNU Terry Pratchett'}}
-
lambda_decorators.
async_handler
(handler)[source]¶ This decorator allows for use of async handlers by automatically running them in an event loop. The loop is added to the context object for if the handler needs it.
Usage:
>>> from lambda_decorators import async_handler >>> async def foobar(): ... return 'foobar' >>> @async_handler ... async def handler(event, context): ... return await foobar() >>> class Context: ... pass >>> handler({}, Context()) 'foobar'
NOTE: Python 3 only
-
lambda_decorators.
before
(func)[source]¶ Run a function before the handler is invoked, is passed the event & context and must return an event & context too.
Usage:
>>> # to create a reusable decorator >>> @before ... def print_request_id(event, context): ... print(context.aws_request_id) ... return event, context >>> @print_request_id ... def handler(event, context): ... pass >>> class Context: ... aws_request_id = 'ID!' >>> handler({}, Context()) ID! >>> # or a one off >>> @before(lambda e, c: (e['body'], c)) ... def handler(body, context): ... return body >>> handler({'body': 'BOOODYY'}, object()) 'BOOODYY'
-
lambda_decorators.
cors_headers
(handler_or_origin=None, origin=None, credentials=False)[source]¶ Automatically injects
Access-Control-Allow-Origin
headers to http responses. Also optionally addsAccess-Control-Allow-Credentials: True
if called withcredentials=True
Usage:
>>> from lambda_decorators import cors_headers >>> @cors_headers ... def hello(example, context): ... return {'body': 'foobar'} >>> hello({}, object()) {'body': 'foobar', 'headers': {'Access-Control-Allow-Origin': '*'}} >>> # or with custom domain >>> @cors_headers(origin='https://example.com', credentials=True) ... def hello_custom_origin(example, context): ... return {'body': 'foobar'} >>> hello_custom_origin({}, object()) {'body': 'foobar', 'headers': {'Access-Control-Allow-Origin': 'https://example.com', 'Access-Control-Allow-Credentials': True}}
-
lambda_decorators.
dump_json_body
(handler_or_none=None, **json_dumps_kwargs)[source]¶ Automatically serialize response bodies with json.dumps.
Returns a 500 error if the response cannot be serialized
Usage:
>>> from lambda_decorators import dump_json_body >>> @dump_json_body ... def handler(event, context): ... return {'statusCode': 200, 'body': {'hello': 'world'}} >>> handler({}, object()) {'statusCode': 200, 'body': '{"hello": "world"}'} >>> from decimal import Decimal >>> @dump_json_body(default=str) ... def handler(event, context): ... return {'statusCode': 200, 'body': {'hello': Decimal('4.2')}} >>> handler({}, object()) {'statusCode': 200, 'body': '{"hello": "4.2"}'}
-
lambda_decorators.
json_http_resp
(handler_or_none=None, **json_dumps_kwargs)[source]¶ Automatically serialize return value to the body of a successfull HTTP response.
Returns a 500 error if the response cannot be serialized
Usage:
>>> from lambda_decorators import json_http_resp >>> @json_http_resp ... def handler(event, context): ... return {'hello': 'world'} >>> handler({}, object()) {'statusCode': 200, 'body': '{"hello": "world"}'} >>> @json_http_resp ... def err_handler(event, context): ... raise Exception('foobar') >>> err_handler({}, object()) {'statusCode': 500, 'body': 'foobar'} >>> from decimal import Decimal >>> @json_http_resp(default=str) ... def handler(event, context): ... return {'hello': Decimal('4.2')} >>> handler({}, object()) {'statusCode': 200, 'body': '{"hello": "4.2"}'}
in this example, the decorated handler returns:
{'statusCode': 200, 'body': '{"hello": "world"}'}
-
lambda_decorators.
json_schema_validator
(request_schema=None, response_schema=None)[source]¶ Validate your request & response payloads against a JSONSchema.
NOTE: depends on the jsonschema package. If youâre using serverless-python-requirements youâre all set. If you cURLed
lambda_decorators.py
youâll have to install it manually in your serviceâs root directory.Usage:
>>> from jsonschema import ValidationError >>> from lambda_decorators import json_schema_validator >>> @json_schema_validator(request_schema={ ... 'type': 'object', 'properties': {'price': {'type': 'number'}}}) ... def handler(event, context): ... return event['price'] >>> handler({'price': 'bar'}, object()) {'statusCode': 400, 'body': "RequestValidationError: 'bar' is not of type 'number'"} >>> @json_schema_validator(response_schema={ ... 'type': 'object', 'properties': {'price': {'type': 'number'}}}) ... def handler(event, context): ... return {'price': 'bar'} >>> handler({}, object()) {'statusCode': 500, 'body': "ResponseValidationError: 'bar' is not of type 'number'"}
-
lambda_decorators.
load_json_body
(handler_or_none=None, **json_loads_kwargs)[source]¶ Automatically deserialize event bodies with json.loads.
Automatically returns a 400 BAD REQUEST if there is an error while parsing.
Usage:
>>> from lambda_decorators import load_json_body >>> @load_json_body ... def handler(event, context): ... return event['body']['foo'] >>> handler({'body': '{"foo": "bar"}'}, object()) 'bar'
note that
event['body']
is already a dictionary and didnât have to explicitly be parsed.
-
lambda_decorators.
load_urlencoded_body
(handler)[source]¶ Automatically deserialize application/x-www-form-urlencoded bodies
Automatically returns a 400 BAD REQUEST if there is an error while parsing.
Usage:
>>> from lambda_decorators import load_urlencoded_body >>> @load_urlencoded_body ... def handler(event, context): ... return event['body']['foo'] >>> handler({'body': 'foo=spam&bar=answer&bar=42'}, object()) ['spam']
note that
event['body']
is already a dictionary and didnât have to explicitly be parsed.
-
lambda_decorators.
no_retry_on_failure
(handler)[source]¶ AWS Lambda retries scheduled lambdas that donât execute succesfully.
This detects this by storing requests IDs in memory and exiting early on duplicates. Since this is in memory, donât use it on very frequently scheduled lambdas. It logs a critical message then exits with a statusCode of 200 to avoid further retries.
Usage:
>>> import logging, sys >>> from lambda_decorators import no_retry_on_failure, logger >>> logger.addHandler(logging.StreamHandler(stream=sys.stdout)) >>> @no_retry_on_failure ... def scheduled_handler(event, context): ... return {'statusCode': 500} >>> class Context: ... aws_request_id = 1 >>> scheduled_handler({}, Context()) {'statusCode': 500} >>> scheduled_handler({}, Context()) Retry attempt on request id 1 detected. {'statusCode': 200}
-
lambda_decorators.
on_exception
(func)[source]¶ Run a function when a handler thows an exception. Itâs return value is returned to AWS.
Usage:
>>> # to create a reusable decorator >>> @on_exception ... def handle_errors(exception): ... print(exception) ... return {'statusCode': 500, 'body': 'uh oh'} >>> @handle_errors ... def handler(event, context): ... raise Exception('it broke!') >>> handler({}, object()) it broke! {'statusCode': 500, 'body': 'uh oh'} >>> # or a one off >>> @on_exception(lambda e: {'statusCode': 500}) ... def handler(body, context): ... raise Exception >>> handler({}, object()) {'statusCode': 500}
-
lambda_decorators.
secret_manager
(secret_name)[source]¶ Get a secret value from the AWS Secret Manager.
Deprecated since version 0.3.
Use the better spelled secrets_manager
-
lambda_decorators.
secrets_manager
(*secret_names)[source]¶ Get secrets from the AWS Secrets Manager.
Secrets are added to a dictionary named
secrets
on the context object.This requires your lambda to have the
secretsmanager:GetSecretValue
permission for the requested secret andkms:Decrypt
for any keys used to encrypt the secrets.Usage:
>>> from lambda_decorators import secrets_manager >>> @secrets_manager('dschep/test') ... def secret_getter(event, context): ... return context.secrets >>> class Context: ... pass >>> secret_getter({}, Context()) {'dschep/test': {'foo': 'b4r', 'floo': 'b4z'}}
-
lambda_decorators.
ssm_parameter_store
(*parameters)[source]¶ Get parameters from the AWS SSM Parameter Store.
Secrets are added to a dictionary named
ssm_params
on the context object.This requires your lambda to have the
ssm:GetParameters
permission on for requested parameters andkms:Decrypt
for any keys used to encrypt the parameters.Usage:
>>> from lambda_decorators import ssm_parameter_store >>> @ssm_parameter_store('/dschep/test') ... def param_getter(event, context): ... return context.parameters >>> class Context: ... pass >>> param_getter({}, Context()) {'/dschep/test': 'f00b4r'}
For more advanced SSM use, see ssm-cache