aws-lambda-decorators
Advanced tools
A set of Python decorators to ease the development of AWS lambda functions.
The easiest way to use these AWS Lambda Decorators is to install them through Pip:
pip install aws-lambda-decorators
The Logging level of the decorators can be controlled by setting a LOG_LEVEL environment variable. In python:
os.environ["LOG_LEVEL"] = "INFO"
The default value is "INFO"
The current list of AWS Lambda Python Decorators includes:
Currently, the package offers 12 validators:
The package offers functions to decode from JSON and JWT.
You can see some basic examples in the examples folder.
This decorator extracts and validates values from dictionary parameters passed to a Lambda Function.
Example:
@extract(parameters=[
Parameter(path='/parent/my_param', func_param_name='a_dictionary'), # extracts a non mandatory my_param from a_dictionary
Parameter(path='/parent/missing_non_mandatory', func_param_name='a_dictionary', default='I am missing'), # extracts a non mandatory missing_non_mandatory from a_dictionary
Parameter(path='/parent/missing_mandatory', func_param_name='a_dictionary'), # does not fail as the parameter is not validated as mandatory
Parameter(path='/parent/child/id', validators=[Mandatory], var_name='user_id', func_param_name='another_dictionary') # extracts a mandatory id as "user_id" from another_dictionary
])
def extract_example(a_dictionary, another_dictionary, my_param='aDefaultValue', missing_non_mandatory='I am missing', missing_mandatory=None, user_id=None):
"""
Given these two dictionaries:
a_dictionary = {
'parent': {
'my_param': 'Hello!'
},
'other': 'other value'
}
another_dictionary = {
'parent': {
'child': {
'id': '123'
}
}
}
you can now access the extracted parameters directly:
"""
return my_param, missing_non_mandatory, missing_mandatory, user_id
Or you can use kwargs instead of specific parameter names:
Example:
@extract(parameters=[
Parameter(path='/parent/my_param', func_param_name='a_dictionary') # extracts a non mandatory my_param from a_dictionary
])
def extract_to_kwargs_example(a_dictionary, **kwargs):
"""
a_dictionary = {
'parent': {
'my_param': 'Hello!'
},
'other': 'other value'
}
"""
return kwargs['my_param'] # returns 'Hello!'
A missing mandatory parameter, or a parameter that fails validation, will raise an exception:
Example:
@extract(parameters=[
Parameter(path='/parent/mandatory_param', func_param_name='a_dictionary', validators=[Mandatory]) # extracts a mandatory mandatory_param from a_dictionary
])
def extract_mandatory_param_example(a_dictionary, mandatory_param=None):
return 'Here!' # this part will never be reached, if the mandatory_param is missing
response = extract_mandatory_param_example({'parent': {'my_param': 'Hello!'}, 'other': 'other value'} )
print(response) # prints { 'statusCode': 400, 'body': '{"message": [{"mandatory_param": ["Missing mandatory value"]}]}' } and logs a more detailed error
You can add custom error messages to all validators, and incorporate to those error messages the validated value and the validation condition:
Example:
@extract(parameters=[
Parameter(path='/parent/an_int', func_param_name='a_dictionary', validators=[Minimum(100, 'Bad value {value}: should be at least {condition}')]) # extracts a mandatory mandatory_param from a_dictionary
])
def extract_minimum_param_with_custom_error_example(a_dictionary, mandatory_param=None):
return 'Here!' # this part will never be reached, if the an_int param is less than 100
response = extract_minimum_param_with_custom_error_example({'parent': {'an_int': 10}})
print(response) # prints { 'statusCode': 400, 'body': '{"message": [{"an_int": ["Bad value 10: should be at least 100"]}]}' } and logs a more detailed error
You can group the validation errors together (instead of exiting on first error).
Example:
@extract(parameters=[
Parameter(path='/parent/mandatory_param', func_param_name='a_dictionary', validators=[Mandatory]), # extracts two mandatory parameters from a_dictionary
Parameter(path='/parent/another_mandatory_param', func_param_name='a_dictionary', validators=[Mandatory]),
Parameter(path='/parent/an_int', func_param_name='a_dictionary', validators=[Maximum(10), Minimum(5)])
], group_errors=True) # groups both errors together
def extract_multiple_param_example(a_dictionary, mandatory_param=None, another_mandatory_param=None, an_int=0):
return 'Here!' # this part will never be reached, if the mandatory_param is missing
response = extract_multiple_param_example({'parent': {'my_param': 'Hello!', 'an_int': 20}, 'other': 'other value'})
print(response) # prints {'statusCode': 400, 'body': '{"message": [{"mandatory_param": ["Missing mandatory value"]}, {"another_mandatory_param": ["Missing mandatory value"]}, {"an_int": ["\'20\' is greater than maximum value \'10\'"]}]}'}
You can decode any part of the parameter path from json or any other existing annotation.
Example:
@extract(parameters=[
Parameter(path='/parent[json]/my_param', func_param_name='a_dictionary') # extracts a non mandatory my_param from a_dictionary
])
def extract_from_json_example(a_dictionary, my_param=None):
"""
a_dictionary = {
'parent': '{"my_param": "Hello!" }',
'other': 'other value'
}
"""
return my_param # returns 'Hello!'
You can also use an integer annotation to access an specific list element by index.
Example:
@extract(parameters=[
Parameter(path='/parent[1]/my_param', func_param_name='a_dictionary') # extracts a non mandatory my_param from a_dictionary
])
def extract_from_list_example(a_dictionary, my_param=None):
"""
a_dictionary = {
'parent': [
{'my_param': 'Hello!'},
{'my_param': 'Bye!'}
]
}
"""
return my_param # returns 'Bye!'
You can extract all parameters into a dictionary
Example:
@extract(parameters=[
Parameter(path='/params/my_param_1', func_param_name='a_dictionary'), # extracts a non mandatory my_param_1 from a_dictionary
Parameter(path='/params/my_param_2', func_param_name='a_dictionary') # extracts a non mandatory my_param_2 from a_dictionary
])
def extract_dictionary_example(a_dictionary, **kwargs):
"""
a_dictionary = {
'params': {
'my_param_1': 'Hello!',
'my_param_2': 'Bye!'
}
}
"""
return kwargs # returns {'my_param_1': 'Hello!', 'my_param_2': 'Bye!'}
You can apply a transformation to an extracted value. The transformation will happen before validation.
Example:
@extract(parameters=[
Parameter(path='/params/my_param', func_param_name='a_dictionary', transform=int) # extracts a non mandatory my_param from a_dictionary
])
def extract_with_transform_example(a_dictionary, my_param=None):
"""
a_dictionary = {
'params': {
'my_param': '2' # the original value is the string '2'
}
}
"""
return my_param # returns the int value 2
The transform function can be any function, with its own error handling.
Example:
def to_int(arg):
try:
return int(arg)
except Exception:
raise Exception("My custom error message")
@extract(parameters=[
Parameter(path='/params/my_param', func_param_name='a_dictionary', transform=to_int) # extracts a non mandatory my_param from a_dictionary
])
def extract_with_custom_transform_example(a_dictionary, my_param=None):
return {}
response = extract_with_custom_transform_example({'params': {'my_param': 'abc'}})
print(response) # prints {'statusCode': 400, 'body': '{"message": "Error extracting parameters"}'}, and the logs will contain the "My custom error message" message.
This decorator is just a facade to the extract method to be used in AWS Api Gateway Lambdas. It automatically extracts from the event lambda parameter.
Example:
@extract_from_event(parameters=[
Parameter(path='/body[json]/my_param', validators=[Mandatory]), # extracts a mandatory my_param from the json body of the event
Parameter(path='/headers/Authorization[jwt]/sub', validators=[Mandatory], var_name='user_id') # extract the mandatory sub value as user_id from the authorization JWT
])
def extract_from_event_example(event, context, my_param=None, user_id=None):
"""
event = {
'body': '{"my_param": "Hello!"}',
'headers': {
'Authorization': 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c'
}
}
"""
return my_param, user_id # returns ('Hello!', '1234567890')
This decorator is just a facade to the extract method to be used in AWS Api Gateway Lambdas. It automatically extracts from the context lambda parameter.
Example:
@extract_from_context(parameters=[
Parameter(path='/parent/my_param', validators=[Mandatory]) # extracts a mandatory my_param from the parent element in context
])
def extract_from_context_example(event, context, my_param=None):
"""
context = {
'parent': {
'my_param': 'Hello!'
}
}
"""
return my_param # returns 'Hello!'
This decorator extracts a parameter from AWS SSM and passes the parameter down to your function as a kwarg.
Example:
@extract_from_ssm(ssm_parameters=[
SSMParameter(ssm_name='one_key'), # extracts the value of one_key from SSM as a kwarg named "one_key"
SSMParameter(ssm_name='another_key', var_name="another") # extracts another_key as a kwarg named "another"
])
def extract_from_ssm_example(your_func_params, one_key=None, another=None):
return your_func_params, one_key, another
This decorator validates a list of non dictionary parameters from your lambda function.
Example:
@validate(parameters=[
ValidatedParameter(func_param_name='a_param', validators=[Mandatory]), # validates a_param as mandatory
ValidatedParameter(func_param_name='another_param', validators=[Mandatory, RegexValidator(r'\d+')]) # validates another_param as mandatory and containing only digits
ValidatedParameter(func_param_name='param_with_schema', validators=[SchemaValidator(Schema({'a': Or(str, dict)}))]) # validates param_with_schema as an object with specified schema
])
def validate_example(a_param, another_param, param_with_schema):
return a_param, another_param, param_with_schema # returns 'Hello!', '123456', {'a': {'b': 'c'}}
validate_example('Hello!', '123456', {'a': {'b': 'c'}})
Given the same function validate_example, a 400 exception is returned if at least one parameter does not validate (as per the extract decorator, you can group errors with the group_errors flag):
validate_example('Hello!', 'ABCD') # returns a 400 status code and an error message
This decorator allows for logging the function arguments and/or the response.
Example:
@log(parameters=True, response=True)
def log_example(parameters):
return 'Done!'
log_example('Hello!') # logs 'Hello!' and 'Done!'
This decorator handles a list of exceptions, returning a 400 response containing the specified friendly message to the caller.
Example:
@handle_exceptions(handlers=[
ExceptionHandler(ClientError, "Your message when a client error happens.")
])
def handle_exceptions_example():
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('non_existing_table')
table.query(KeyConditionExpression=Key('user_id').eq(user_id))
# ...
handle_exceptions_example() # returns {'body': '{"message": "Your message when a client error happens."}', 'statusCode': 400}
This decorator handles all exceptions thrown by a lambda, returning a 400 response and the exception's message.
Example:
@handle_all_exceptions()
def handle_exceptions_example():
test_list = [1, 2, 3]
invalid_value = test_list[5]
# ...
handle_all_exceptions_example() # returns {'body': '{"message": "list index out of range"}, 'statusCode': 400}
This decorator ensures that, if the response contains a body, the body is dumped as json.
Example:
@response_body_as_json
def response_body_as_json_example():
return {'statusCode': 400, 'body': {'param': 'hello!'}}
response_body_as_json_example() # returns { 'statusCode': 400, 'body': "{'param': 'hello!'}" }
This decorator adds your defined CORS headers to the decorated function response.
Example:
@cors(allow_origin='*', allow_methods='POST', allow_headers='Content-Type', max_age=86400)
def cors_example():
return {'statusCode': 200}
cors_example() # returns {'statusCode': 200, 'headers': {'access-control-allow-origin': '*', 'access-control-allow-methods': 'POST', 'access-control-allow-headers': 'Content-Type', 'access-control-max-age': 86400}}
This decorator adds HSTS header to the decorated function response. Uses 2 years max-age (recommended default from https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Strict-Transport-Security) unless custom value provided as parameter.
Example:
@hsts()
def hsts_example():
return {'statusCode': 200}
hsts_example() # returns {'statusCode': 200, 'headers': {'Strict-Transport-Security': 'max-age=63072000'}}
This decorator pushes unsuccessful responses back to the calling client over websockets built on api gateway
This decorator requires the client is connected to the websocket api gateway instance, and will therefore have a connection id
Example:
@push_ws_errors('https://api_id.execute_id.region.amazonaws.com/Prod')
@handle_all_exceptions()
def handler(event, context):
return {
'statusCode': 400,
'body': {
'message': 'Bad request'
}
}
# will push {'type': 'error', 'statusCode': 400, 'message': 'Bad request'} back to the client via websockets
This decorator pushes all responses back to the calling client over websockets built on api gateway
This decorator requires the client is connected to the websocket api gateway instance, and will therefore have a connection id
Example:
@push_ws_response('https://api_id.execute_id.region.amazonaws.com/Prod')
def handler(event, context):
return {
'statusCode': 200,
'body': 'Hello, world!'
}
# will push {'statusCode': 200, 'body': 'Hello, world!'} back to the client via websockets
You can create your own validators by inheriting from the Validator class.
Fix length validator example:
class FixLength(Validator):
ERROR_MESSAGE = "'{value}' length should be '{condition}'"
def __init__(self, fix_length: int, error_message=None):
super().__init__(error_message=error_message, condition=fix_length)
def validate(self, value=None):
if value is None:
return True
return len(str(value)) == self._condition
You can get the docstring help by running:
>>> from aws_lambda_decorators.decorators import extract
>>> help(extract)
FAQs
A set of python decorators to simplify aws python lambda development
We found that aws-lambda-decorators demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 3 open source maintainers collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
In this segment of the Risky Business podcast, Feross Aboukhadijeh and Patrick Gray discuss the challenges of tracking malware discovered in open source softare.
Research
Security News
A threat actor's playbook for exploiting the npm ecosystem was exposed on the dark web, detailing how to build a blockchain-powered botnet.
Security News
NVD’s backlog surpasses 20,000 CVEs as analysis slows and NIST announces new system updates to address ongoing delays.