Jeffy is Serverless Application Framework for Python AWS Lambda.
Jeffy is Serverless "Application" Framework for Python, which is suite of Utilities for Lambda functions to make it easy to develop serverless applications.
Mainly, Jeffy is focusing on three things.
- Logging: Providing easy to see JSON format logging. All decorators are capturing all events, responses and errors. And you can configure to inject additional attributes what you want to see to logs.
- Decorators: To save time to implement common things for Lambda functions, providing some useful decorators and utiliies.
- Tracing: Traceable events within related functions and AWS services with generating and passing
correlation_id. - Configurable: You can customize the framework settings easily.
-
- 2.1. common
- 2.2. rest_api
- 2.3. sqs
- 2.4. sqs_raw
- 2.5. sns
- 2.6. sns_raw
- 2.7. kinesis_streams
- 2.8. kinesis_streams_raw
- 2.9. dynamodb_streams
- 2.10. s3
- 2.11. schedule
-
- 3.1. Kinesis Clinent
- 3.2. SNS Client
- 3.3. SQS Client
- 3.4. S3 Client
-
- 5.1. JSONSchemaValidator
$ pip install jeffyJeffy logger automatically inject some Lambda contexts to CloudWatchLogs.
from jeffy.framework import get_app
app = get_app()
def handler(event, context):
app.logger.info({'foo': 'bar'})Output in CloudWatchLogs
{
"msg": {"foo":"bar"},
"aws_region":"us-east-1",
"function_name":"jeffy-dev-hello",
"function_version":"$LATEST",
"function_memory_size":"1024",
"log_group_name":"/aws/lambda/jeffy-dev-hello",
"log_stream_name":"2020/01/21/[$LATEST]d7729c0ea59a4939abb51180cda859bf",
"correlation_id":"f79759e3-0e37-4137-b536-ee9a94cd4f52"
}You can inject some additional attributes what you want to output with using update_context method.
from jeffy.framework import get_app
app = get_app()
app.logger.update_context({
'username': 'user1',
'email': '[email protected]'
})
def handler(event, context):
app.logger.info({'foo': 'bar'})Output in CloudWatchLogs
{
"msg": {"foo":"bar"},
"username":"user1",
"email":"[email protected]",
"aws_region":"us-east-1",
"function_name":"jeffy-dev-hello",
"function_version":"$LATEST",
"function_memory_size":"1024",
"log_group_name":"/aws/lambda/jeffy-dev-hello",
"log_stream_name":"2020/01/21/[$LATEST]d7729c0ea59a4939abb51180cda859bf",
"correlation_id":"f79759e3-0e37-4137-b536-ee9a94cd4f52"
}You can change the attribute name of correlation id in the setting options.
from jeffy.framework import get_app
from jeffy.settings import Logging
app = get_app(logging=Logging(correlation_attr_name='my-trace-id'))
def handler(event, context):
app.logger.info({'foo': 'bar'})Output in CloudWatchLogs
{
"msg": {"foo":"bar"},
"aws_region":"us-east-1",
"function_name":"jeffy-dev-hello",
"function_version":"$LATEST",
"function_memory_size":"1024",
"log_group_name":"/aws/lambda/jeffy-dev-hello",
"log_stream_name":"2020/01/21/[$LATEST]d7729c0ea59a4939abb51180cda859bf",
"my-trace-id":"f79759e3-0e37-4137-b536-ee9a94cd4f52"
}You can change the log level of Jeffy logger.
import logging
from jeffy.framework import get_app
from jeffy.settings import Logging
app = get_app(logging=Logging(log_level=logging.DEBUG))
def handler(event, context):
app.logger.info({'foo': 'bar'})Decorators make simple to implement common lamdba tasks, such as parsing array from Kinesis, SNS, SQS events etc.
Here are provided decorators
common decorator allows you to output event, response and error infomations when you face Exceptions
from jeffy.framework import get_app
app = get_app()
app.logger.update_context({
'username': 'user1',
'email': '[email protected]'
})
@app.handlers.common()
def handler(event, context):
...Error output with auto_logging
{
"msg": "JSONDecodeError('Expecting value: line 1 column 1 (char 0)')",
"exec_info":"Traceback (most recent call last):
File '/var/task/jeffy/decorators.py', line 41, in wrapper
raise e
File '/var/task/jeffy/decorators.py', line 36, in wrapper
result = func(event, context)
File '/var/task/handler.py', line 8, in hello
json.loads('s')
File '/var/lang/lib/python3.8/json/__init__.py', line 357, in loads
return _default_decoder.decode(s)
File '/var/lang/lib/python3.8/json/decoder.py', line 337, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
File '/var/lang/lib/python3.8/json/decoder.py', line 355, in raw_decode
raise JSONDecodeError('Expecting value', s, err.value) from None",
"function_name":"jeffy-dev-hello",
"function_version":"$LATEST",
"function_memory_size":"1024",
"log_group_name":"/aws/lambda/jeffy-dev-hello",
"log_stream_name":"2020/01/21/[$LATEST]90e1f70f6e774e07b681e704646feec0",
"correlation_id":"f79759e3-0e37-4137-b536-ee9a94cd4f52"
}
Decorator for API Gateway event. Automatically get the correlation id from request header and set the correlation id to response header.
Default encoding is jeffy.encoding.json.JsonEncoding.
from jeffy.framework import get_app
app = get_app()
@app.handlers.rest_api()
def handler(event, context):
return {
'statusCode': 200,
'headers': 'Content-Type':'application/json',
'body': json.dumps({
'resutl': 'ok.'
})
}Default header name is 'x-jeffy-correlation-id'. You can change this name in the setting option.
from jeffy.framework import get_app
from jeffy.settings import RestApi
app = get_app(
rest_api=RestApi(correlation_id_header='x-foo-bar'))
@app.handlers.rest_api()
def handler(event, context):
return {
'statusCode': 200,
'headers': 'Content-Type':'application/json',
'body': json.dumps({
'resutl': 'ok.'
})
}Decorator for sqs event. Automaticlly parse "event.Records" list from SQS event source to each items for making it easy to treat it inside main process of Lambda.
Default encoding is jeffy.encoding.json.JsonEncoding.
from jeffy.framework import get_app
app = get_app()
@app.handlers.sqs()
def handler(event, context):
return event['foo']
"""
"event.Records" list from SQS event source was parsed each items
if event.Records value is the following,
[
{'foo': 1},
{'foo': 2}
]
event['foo'] value is 1 and 2, event['correlation_id'] is correlation_id you should pass to next event
"""Decorator for sqs raw event (with all metadatas). Automaticlly parse "event.Records" list from SQS event source and pass the each records to main process of Lambda.
Default encoding is jeffy.encoding.json.JsonEncoding.
from jeffy.framework import get_app
app = get_app()
@app.handlers.sqs_raw()
def handler(event, context):
return event['body']Decorator for sns event. Automaticlly parse event.Records list from SNS event source to each items for making it easy to treat it inside main process of Lambda.
Default encoding is jeffy.encoding.json.JsonEncoding.
from jeffy.framework import get_app
app = get_app()
@app.handlers.sns()
def handler(event, context):
return event['foo']
"""
"event.Records" list from SNS event source was parsed each items
if event.Records value is the following,
[
{'foo': 1},
{'foo': 2}
]
event['foo'] value is 1 and 2, event['correlation_id'] is correlation_id you should pass to next event
"""Decorator for sqs raw event (with all metadatas). Automaticlly parse "event.Records" list from SNS event source and pass the each records to main process of Lambda.
Default encoding is jeffy.encoding.json.JsonEncoding.
from jeffy.framework import get_app
app = get_app()
@app.handlers.sns_raw()
def handler(event, context):
return event['Sns']['Message']Decorator for kinesis stream event. Automaticlly parse event.Records list from Kinesis event source to each items and decode it with base64 for making it easy to treat it inside main process of Lambda.
Default encoding is jeffy.encoding.json.JsonEncoding.
from jeffy.framework import get_app
app = get_app()
@app.handlers.kinesis_streams()
def handler(event, context):
return event['foo']
"""
"event.Records" list from Kinesis event source was parsed each items
and decoded with base64 if event.Records value is the following,
[
<base64 encoded value>,
<base64 encoded value>
]
event['foo'] value is 1 and 2, event['correlation_id'] is correlation_id you should pass to next event
"""Decorator for sqs raw event (with all metadatas). Automaticlly parse "event.Records" list from Kinesis Data Streams event source and pass the each records to main process of Lambda.
Default encoding is jeffy.encoding.json.JsonEncoding.
from jeffy.framework import get_app
app = get_app()
@app.handlers.sns_raw()
def handler(event, context):
return event['kinesis']['data']Decorator for dynamodb stream event. Automaticlly parse event.Records list from Dynamodb event source to items for making it easy to treat it inside main process of Lambda.
from jeffy.framework import get_app
app = get_app()
@app.handlers.dynamodb_streams()
def handler(event, context):
return event['foo']
"""
"event.Records" list from Dynamodb event source was parsed each items
if event.Records value is the following,
[
{'foo': 1},
{'foo': 2}
]
event['foo'] value is 1 and 2, event['correlation_id'] is correlation_id you should pass to next event
"""Decorator for S3 event. Automatically parse body stream from triggered S3 object and S3 bucket and key name to Lambda.
This handler requires s3:GetObject permission.
Default encoding is jeffy.encoding.bytes.BytesEncoding.
from jeffy.framework import get_app
app = get_app()
@app.handlers.s3()
def handler(event, context):
event['key'] # S3 bucket key
event['bucket_name'] # S3 bucket name
event['body'] # Bytes data of the object
event['correlation_id'] # correlation_id
event['metadata'] # object matadataDecorator for schedule event. just captures correlation id before main Lambda process. do nothing other than that.
from jeffy.framework import setup
app = setup()
@app.handlers.schedule()
def handler(event, context):
...Jeffy has the original wrapper clients of AWS SDK(boto3). The clients automatically inject correlation_id in the event payload and encode it to the specified(or default) encoding.
Default encoding is jeffy.encoding.json.JsonEncoding.
from jeffy.framework import get_app
from jeffy.sdk.kinesis import Kinesis
app = get_app()
@app.handlers.kinesis_streams()
def handler(event, context):
Kinesis().put_record(
stream_name=os.environ['STREAM_NAME'],
data={'foo': 'bar'},
partition_key='your_partition_key'
)Default encoding is jeffy.encoding.json.JsonEncoding.
from jeffy.framework import get_app
from jeffy.sdk.sns import Sns
app = get_app()
@app.handlers.sns()
def handler(event, context):
Sns().publish(
topic_arn=os.environ['TOPIC_ARN'],
subject='hello',
message='world'
)Default encoding is jeffy.encoding.json.JsonEncoding.
Jeffy wraps the 'message' argument into JSON data with tracing ID attributes then sends it to the queue.
from jeffy.framework import get_app
from jeffy.sdk.sqs import Sqs
app = get_app()
@app.handlers.sqs()
def handler(event, context):
Sqs().send_message(
queue_url=os.environ['QUEUE_URL'],
message='hello world'
)In the above sample, your 'hello world' is sent to the queue as the following JSON data (by default, jeffy appends 'correlation_id' as the tracing ID attribute).
{
"message": "hello world",
"correlation_id": "<auto-generated-uuid>"
}You are also able to use dict value as the message. You are also able to use dict type as the message. Jeffy will insert your dict value to 'message' attribute.
Note: It recommended use default encoder, JsonEncoder. You can take not use to JsonEncoder but you need to implements Custom Encoder is able to encode dict types (currently other encoder, BytesEncoder Jeffy provided is not support this use case).
Default encoding is jeffy.encoding.bytes.BytesEncoding.
from jeffy.framework import get_app
from jeffy.sdk.s3 import S3
app = get_app()
@app.handlers.s3()
def handler(event, context):
S3().upload_file(
file_path='/path/to/file',
bucket_name=os.environ['BUCKET_NAME'],
key='key/of/object'
)Jeffy encoder works as a type in order to consistent the specification of messages between AWS Lambda and other AWS services. This allows you to implement Lambda function more safely, efficiently and solidly by defining the type of messages from the begging to the end of the microservices.
Each handler and SDK client has a default encoding and automatically encode/decode the data from/to python object. And you can change the encoding.
Currently, the encodings you can choose are:
jeffy.encoding.bytes.BytesEncoding: For using binary data.jeffy.encoding.json.JsonEncoding: For using JSON format. At this time, Jeffy uses this as a default encoder except for Amazon S3.
Jeffy internally uses AWS SDK for python which is a wrapper of boto3. When publishing the message to AWS Services from Lambda function, Jeffy encodes automatically with a default encoder. On the other hand, when AWS Lambda subscribes to messages, Jeffy decodes automatically with the encoder of each AWS Services that is the event source. After thet, you can handle messages inside of Lambda function.
Each encoding class also has encode methods to encode bytes data into own encoding. Thus the developer can set a specific encoder you want to use.
from jeffy.framework import get_app
from jeffy.encoding.bytes import BytesEncoding
from jeffy.sdk.kinesis import Kinesis
app = get_app()
bytes_encoding = BytesEncoding()
@app.handlers.kinesis_streams(encoding=bytes_encoding)
def handler(event, context):
kinesis = Kinesis(encoding=bytes_encoding)
kinesis.put_record(
stream_name=os.environ['STREAM_NAME'],
data=bytes_encoding.encode('foo'.encode('utf-8)),
partition_key='your-partition-key'
)This encoding mechanism simplifies to consistent the specification of messages of the event-driven architecture which is consist of Pub - Sub style.
JsonSchemaValidator is automatically validate event payload with following json schema you define. raise ValidationError exception if the validation fails.
from jeffy.framework import get_app
from jeffy.validator.jsonschema import JsonSchemaValidator
app = get_app()
@app.handlers.rest_api(
validator=JsonSchemaValidator(schema={
'type': 'object',
'properties': {
'message': {'type': 'string'}}}))
def handler(event, context):
return {
'statusCode': 200,
'headers': 'Content-Type':'application/json',
'body': json.dumps({
'message': 'ok.'
})
}- Python 3.6 or higher
- Source hosted at GitHub
- Report issues/questions/feature requests on GitHub Issues
Pull requests are very welcome! Make sure your patches are well tested. Ideally create a topic branch for every separate change you make. For example:
- Fork the repo
- Create your feature branch (
git checkout -b my-new-feature) - Commit your changes (
git commit -am"Added some feature") - Push to the branch (
git push origin my-new-feature) - Create new Pull Request
- Bought up initial idea by Masashi Terui ([email protected])
- Created and maintained by Serverless Operations, Inc
Jeffy is inspired by the following products.
MIT License (see LICENSE)
Jeffy has many users who can speak Japanese as their first language. You can give feedback and discuss about Jeffy in Japanese here if you want.