Syngenta Digital ALC (AWS Lambda Client)

Auto-loading, self-validating, minimalist python framework for AWS Lambdas

Features

• Automatic routing based on folder structure
• 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 amazon lambdas.

The alc encourages you to use small, internally routed API lambdas in a normalized OOP way.

In addition, it makes things 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

or

$ pipenv install syngenta_digital_alc

Basic Usage

apigateway events

NOTE: This packages assumes you are using a monolithic, internally routed lambda with serverless framework and you have a folder structure which puts all your endpoint files in one sub directory tied to a custom domain on apigateway.

0. 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    

1. 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)

2. 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

Request Properties

Response Properties

(OPTIONAL) RUN LOGIC BEFORE EVERY REQUEST

This is a feature which allows you to interrogate all requests before they hit your endpoint. Here are some things to remember:

• your function will run only after a valid route and method have been determined
• runs before any validation
• requires you use the BeforeAllException class to stop processing or route will continue
• must be a pure function that is passed in context

Example configuration

import os

from syngenta_digital_alc.apigateway.router import Router

from example.function.to.import import example_before_all


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,
        before_all=example_before_all.run # this is the important part
    )
    return router.route()   

Example before all function

from syngenta_digital_alc.apigateway.custom_exceptions import BeforeAllException
from syngenta_digital_alc.apigateway.handler_requirements import handler_requirements


@handler_requirements()
def run(request, response):
    if not request.headers.get('x-api-key') == 'some-secret-key':
        raise BeforeAllException(code=401, key_path='headers:x-api-key', message='you need an api key')

sqs events

0. 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' ]

1. 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

Record Properties

dynamodb events

0. Setting Up your lambda to listen to the dynamodb streams

functions:
    v1-dynamodb-stream:
        name: v1-dynamodb-stream
        handler: application.v1.handler.dynamodb.streamer.stream
        events:
            - stream:
                type: dynamodb
                arn:
                    Fn::GetAtt: [ DynamoDbTableExample, 'Arn' ]

1. Initialize the Event and Iterate over the Records

from syngenta_digital_alc.dynamodb.handler_requirements import handler_requirements

@handler_requirements()
def handle_ddb_trigger(event):
    records = event.records
    for ddb_record in records:
        some_func(ddb_record)

Event Client Properties

Record Properties

s3 events

0. Setting Up your lambda to listen to s3 bucket changes

functions:
    v1-s3-handler:
        name: v1-s3-handler
        handler: application.v1.handler.s3.handler.handle
        events:
            - s3: photos

1. Initialize the Event and Iterate over the Records

from syngenta_digital_alc.s3.handler_requirements import handler_requirements

@handler_requirements()
def handle_s3_trigger(event):
    records = event.records
    for s3_record in records:
        some_func(s3_record)

Event Client Properties

Record Properties

generic events

0. Setting Up your lambda to listen to take in random events

functions:
    v1-generic-handler:
        name: v1-generic-handler
        handler: application.v1.handler.console.handler.handle

1. Initialize the Event and Iterate over the Records

from syngenta_digital_alc.generic.handler_requirements import handler_requirements

@handler_requirements()
def handle_generic_trigger(event):
    some_func(event.body) # will automatically decode JSON, if it is JSON

Event Client Properties

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:

pipenv run test