A set of python decorators to simplify aws python lambda development
Project description
aws-lambda-decorators
A set of Python decorators to ease the development of AWS lambda functions.
Installation
The easiest way to use these AWS Lambda Decorators is to install them through Pip:
pip install aws-lambda-decorators
Logging
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"
Package Contents
Decorators
The current list of AWS Lambda Python Decorators includes:
- extract: a decorator to extract and validate specific keys of a dictionary parameter passed to a AWS Lambda function.
- extract_from_event: a facade of extract to extract and validate keys from an AWS API Gateway lambda function event parameter.
- extract_from_context: a facade of extract to extract and validate keys from an AWS API Gateway lambda function context parameter.
- extract_from_ssm: a decorator to extract from AWS SSM the values of a set of parameter keys.
- validate: a decorator to validate a list of function parameters.
- log: a decorator to log the parameters passed to the lambda function and/or the response of the lambda function.
- handle_exceptions: a decorator to handle any type of declared exception generated by the lambda function.
- response_body_as_json: a decorator to transform a response dictionary body to a json string.
- handle_all_exceptions: a decorator to handle all exceptions thrown by the lambda function.
- cors: a decorator to add cors headers to a lambda function.
Validators
Currently, the package offers 5 validators:
- Mandatory: Checks if a parameter has a value.
- RegexValidator: Checks a parameter against a regular expression.
- SchemaValidator: Checks if an object adheres to the schema. Uses schema library.
- Minimum: Checks if an optional numerical value is greater than a minimum value.
- Maximum: Checks if an optional numerical value is less than a maximum value.
- MinLength: Checks if an optional string value is longer than a minimum length.
- MaxLength: Checks if an optional string value is shorter than a maximum length.
- Type: Checks if an optional object value is of a given python type.
- EnumValidator: Checks if an optional object value is in a list of valid values.
- NonEmpty: Checks if an optional object value is not an empty value.
- DateValidator: Checks if a given string is a valid date according to a passed in date format.
Decoders
The package offers functions to decode from JSON and JWT.
- decode_json: decodes/converts a json string to a python dictionary
- decode_jwt: decodes/converts a JWT string to a python dictionary
Examples
extract
This decorator extracts and validates values from dictionary parameters passed to a Lambda Function.
- The decorator takes a list of Parameter objects.
- Each Parameter object requires a non-empty path to the parameter in the dictionary, and the name of the dictionary (func_param_name)
- The parameter value is extracted and added as a kwarg to the lambda handler (or any other decorated function/method).
- You can add the parameter to the handler signature, or access it in the handler through kwargs.
- The name of the extracted parameter is defaulted to the last element of the path name, but can be changed by passing a (valid pythonic variable name) var_name
- You can define a default value for the parameter in the Parameter or in the lambda handler itself.
- A 400 exception is raised when the parameter cannot be extracted or when it does not validate.
- A variable path (e.g. '/headers/Authorization[jwt]/sub') can be annotated to specify a decoding. In the example, Authorization might contain a JWT, which needs to be decoded before accessing the "sub" element.
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.
extract_from_event
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')
extract_from_context
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!'
extract_from_ssm
This decorator extracts a parameter from AWS SSM and passes the parameter down to your function as a kwarg.
- The decorator takes a list of SSMParameter objects.
- Each SSMParameter object requires the name of the SSM parameter (ssm_name)
- If no var_name is passed in, the extracted value is passed to the function with the ssm_name name
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
validate
This decorator validates a list of non dictionary parameters from your lambda function.
- The decorator takes a list of ValidatedParameter objects.
- Each parameter object needs the name of the lambda function parameter that it is going to be validated, and the list of rules to validate.
- A 400 exception is raised when the parameter does not validate.
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
log
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!'
handle_exceptions
This decorator handles a list of exceptions, returning a 400 response containing the specified friendly message to the caller.
- The decorator takes a list of ExceptionHandler objects.
- Each ExceptionHandler requires the type of exception to check, and an optional friendly message to return 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}
handle_all_exceptions
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}
response_body_as_json
This decorator ensures that, if the response contains a body, the body is dumped as json.
- Returns a 500 error if the response body cannot be 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!'}" }
cors
This decorator adds your defined CORS headers to the decorated function response.
- Returns a 500 error if one or more of the CORS headers have an invalid type
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}}
Writing your own validators
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
Documentation
You can get the docstring help by running:
>>> from aws_lambda_decorators.decorators import extract
>>> help(extract)
Links
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for aws-lambda-decorators-0.45.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | c24f312039c29b53af94958c8c8dc6447fb67e02fe86182331be2ba94ae17799 |
|
MD5 | 7fad8a9e1fa79a5528f90ceb0f9754d7 |
|
BLAKE2b-256 | 5527ff02e04d02d5061c432c7ddffa17ee538e9abe74134d381a78428ebf389c |
Hashes for aws_lambda_decorators-0.45-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4b5094561868c4118b0bf402e6524f28e52ecc1b17d411c9502b29edb1b7f9ac |
|
MD5 | edcca345422bea0da7e77d5b65953a72 |
|
BLAKE2b-256 | 0115445da605c683a96fd5a0f686bb4fbdd0dd246d09863f10968c54e0094b2b |