DRY approach to working with AWS Lambdas
Project description
alc-python
Auto-loading, self-validating, minimalist python framework for Amazon Web Service Lambdas
Features
- Automatic routing based on folder structure
- Focus remove unnecessary boilerplate from your development process
- Ease-of-use with the serverless framework
- Local Development support
Philosophy
The alc philosophy is to provide a self evident tool for use with the amazon lambdas.
The alc encourages you to use small, internally routed API lambdas in a normalized OOP way.
In addition, it makes this like routing and validating API requests less cumbersome and time consuming.
Installation
This is a python module available through the pypi registry.
Before installing, download and install python. python 3 or higher is required.
Installation is done using the
pip install
command:
$ pip install syngenta_digital_alc
Basic Usage
Use alc to normalize your interactions with common events passed to AWS Lambdas and work with the events in a more OO way.
apigateway events
NOTE
: This packages assumes you are using a monolithic, internally routed the lambda with serverless framework and folder structure which put all your endpoint files in one sub directory and custom domain on apigateway.
- Setting Up the Monolithic Internally Routed Lambda
functions:
v1-apigateway-handler:
handler: application.v1.handler.apigateway._router.route
events:
- http:
path: /v1/
method: ANY
- http:
path: /v1/{proxy+}
method: ANY
- Initialize the Router
import os
from syngenta_digital_alc.apigateway.router import Router
# must pass current service, version of API and where handlers are located
def _route(event, context):
router = Router(
base_path='{}/{}'.format(os.environ['service'], 'v1'),
handler_path='application.v1.controller.apigateway',
schema_path='application/openapi.yml',
event=event,
context=context
)
return router.route()
# examples of how router routes (called route -> will import):
# api.url.com/service/v1 -> application.v1.handler.apigateway.__init__.py
# api.url.com/service/v1/hello -> application.v1.handler.apigateway.hello.py
# api.url.com/service/v1/hello-world -> application.v1.handler.apigateway.hello_world.py
# @NOTE: router will always match HTTP Method with function name
# @NOTE: router does NOT support path parameters (please use query strings)
Option Name | Required | Type | Default | Description |
---|---|---|---|---|
event |
true | dict | n/a | event object passed into lambda |
base_path |
true | string | n/a | apigateway base path for the custom domain |
handler_path |
true | string | n/a | project path of where the endpoint files |
schema_path |
false | string | null | path where your schema file can found (accepts JSON as well) |
before_all |
false | string | null | before all middleware function to run before all routes (after validation occurs) |
after_all |
false | string | null | after all middleware function to run after all routes |
- Create handler file with matching methods and requirements
from syngenta_digital_alc.apigateway.handler_requirements import handler_requirements
@handler_requirements(
required_headers=['x-login-token', 'x-permission-token'],
available_headers=['id', 'overwrite'],
required_params=['id', 'overwrite'],
available_params=['id', 'overwrite'],
required_body='v1-post-example-request'
)
def post(request, response):
response.body = business_layer.some_function(request.body)
response.code = 201
Endpoint Requirement Options
Option Name | Type | Description |
---|---|---|
required_body |
string | the components schema key name |
available_params |
list | list of available query string params for that method on that endpoint |
required_params |
list | list of required query string params for that method on that endpoint |
available_headers |
list | list of available headers for that method on that endpoint |
required_headers |
list | list of required headers for that method on that endpoint |
Request Properties
Property Name | Description |
---|---|
method |
method id of apigateway event |
resource |
resource handle of apigateway event |
authorizer |
authorizer of apigateway event (will default to use headers if using with serverless offline) |
headers |
headers of apigateway event |
params |
query string params of apigateway event |
path |
path arguments of apigateway event |
json |
body of apigateway event parsed as JSON |
xml |
body of apigateway event parsed as XML |
body |
body of apigateway event will be parsed based on context headers |
request |
full request broken down as an object literal |
Response Properties
Property Name | Description |
---|---|
headers |
headers you want to send in response |
code |
status code of response (will default to 204 if no content && will default 400 if errors found in response) |
authorizer |
authorizer of apigateway event (will default to use headers if using with serverless offline) |
has_errors() |
function will tell you if errors in the response |
set_error() |
function will set error key and message |
sqs events
- Setting Up your lambda to listen to the Queue
functions:
v1-sqs-subscription:
name: v1-sqs-subscription
handler: application.v1.handler.sqs.listener.listen
events:
- sqs:
arn:
Fn::GetAtt: [ ExampleQueue, 'Arn' ]
- Initialize the Event and Iterate over the Records
from syngenta_digital_alc.sqs.handler_requirements import handler_requirements
@handler_requirements()
def handle_sqs_trigger(event):
records = event.records
for sqs_record in records:
some_func(sqs_record)
Event Client Properties
Property Name | Description |
---|---|
records |
list of record objects |
raw_records |
jus the raw record from the original request |
Record Properties
Property Name | Description |
---|---|
message_id |
message id of sqs record |
receipt_handle |
receipt handle of sqs record |
body |
body of sqs record (will automatically decode JSON) |
raw_body |
body of sqs record |
attributes |
attributes of sqs record |
message_attributes |
message attributes of sqs record |
md5_of_body |
md5 of body of sqs record |
source |
source of sqs record |
source_arn |
source ARN of sqs record |
region |
region of sqs record |
dynamodb events
- Setting Up your lambda to listen to the Queue
functions:
v1-dynamodb-stream:
name: v1-dynamodb-stream
handler: application.v1.handler.dynamodb.streamer.stream
events:
- stream:
type: dynamodb
arn:
Fn::GetAtt: [ DynamoDbTableExample, 'Arn' ]
- Initialize the Event and Iterate over the Records
from syngenta_digital_alc.dynamodb.handler_requirements import handler_requirements
@handler_requirements()
def handle_sqs_trigger(event):
records = event.records
for sqs_record in records:
some_func(sqs_record)
Event Client Properties
Property Name | Description |
---|---|
records |
list of record objects |
raw_records |
jus the raw record from the original request |
Record Properties
Property Name | Description |
---|---|
event_id |
event id of dynamodb record |
event_name |
event name of dynamodb record |
event_source |
event source of dynamodb record |
keys |
keys of dynamodb record (will convert ddb json) |
old_image |
old image of dynamodb record |
new_image |
new image of dynamodb record |
raw_body |
raw of body of dynamodb record |
event_source_arn |
event source ARN of dynamodb record |
event_version |
event version of dynamodb record |
stream_view_type |
stream view type version of dynamodb record |
size_bytes |
size bytes of dynamodb record |
approximate_creation_datetime |
approximate creation date time of dynamodb record |
s3 events
- Setting Up your lambda to listen to the Queue
functions:
v1-s3-handler:
name: v1-s3-handler
handler: application.v1.handler.s3.handler.handle
events:
- s3: photos
- Initialize the Event and Iterate over the Records
from syngenta_digital_alc.s3.handler_requirements import handler_requirements
@handler_requirements()
def handle_sqs_trigger(event):
records = event.records
for sqs_record in records:
some_func(sqs_record)
Event Client Properties
Property Name | Description |
---|---|
records |
list of record objects |
raw_records |
jus the raw record from the original request |
Record Properties
Property Name | Description |
---|---|
event_time |
event time of s3 record |
event_name |
event name of s3 record |
event_source |
event source of s3 record |
region |
region of s3 record (will convert ddb json) |
request_parameters |
request parameters of s3 record |
response_elements |
response elements of s3 record |
configuration_id |
configuration id of s3 record |
object |
object of s3 record |
bucket |
bucket of s3 record |
s3_schema_version |
s3 schema version of s3 record |
Contributing
If you would like to contribute please make sure to follow the established patterns and unit test your code:
Unit Testing
To run unit test, enter command:
RUN_MODE=unittest python -m unittest discover
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 syngenta_digital_alc-1.0.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | d23355205f305f30ba6635890aa197b3cd0095ea7b6418834553408a94125b3d |
|
MD5 | bc5b2e1df83bd72e190b56d1cde519a1 |
|
BLAKE2b-256 | ad1eb4a4658a9f100b5aa499039311349914f3245392e4926ba8bcad4f99f79e |
Hashes for syngenta_digital_alc-1.0.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | b01c645a7985c216bc08d2349a0fac70adef70748ebef6960a1fe877519e2b65 |
|
MD5 | 2670b06090a827e6c0683920ae0b861e |
|
BLAKE2b-256 | e1dcb39a09be9084955c3834b5e6df5ce7b1f98114db1bfd900c4934c2c683f5 |