広告に関する明記

当ブログにはアフィリエイトリンクが含まれます。

AWS IT Jeffy Lambda Python プログラム

jeffyのGitHubページにある英文の和訳

https://github.com/serverless-operations/jeffy
上記ページにある英文の和訳。翻訳ツールにかけただけの機械翻訳ね。
和製なんだから日本語Verもあれば良いのに。

以下、訳文。

説明
Jeffyはサーバーレスの "アプリケーション "フレームワーク for Pythonで、
Lambda関数でのサーバーレスアプリケーション開発を容易にするためのユーティリティのスイートです。

主に、Jeffyは3つのことに力を入れています。
・ロギング：
見やすいJSON形式のロギングを提供します。すべてのデコレータは、すべてのイベント、レスポンス、エラーを取得しています。
そして、あなたがログに見たいものを追加の属性を注入するように設定することができます。
・デコレータ：
Lambda関数でよく使われる処理を実装する手間を節約するために、いくつかの便利なデコレータとユーティリティを提供します。
・トレーシング：
関連する関数やAWSサービス内のイベントを、correlation_idを生成して渡すことでトレース可能にする。

・設定：
フレームワークの設定を簡単にカスタマイズできます。

※目次は略

Install

pip install jeffy

pipコマンドでインストールできます。
特徴
1. Logging　ログの記録
1.1. Basic Usage 基本的な使い方
Jeffy logger は CloudWatchLogs にいくつかの Lambda コンテキストを自動的に注入します。

from jeffy.framework import get_app

app = get_app()

def handler(event, context):
    app.logger.info({'foo': 'bar'})

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"
}

1.2. Injecting additional attributes to logs　ログへの追加属性の注入
update_contextメソッドを使って、出力したい属性を追加で注入することができます。

from jeffy.framework import get_app
app = get_app()

app.logger.update_context({
   'username': 'user1',
   'email': 'user1@example.com'
})

def handler(event, context):
    app.logger.info({'foo': 'bar'})

CloudWatchLogsでの出力

{
   "msg": {"foo":"bar"},
   "username":"user1",
   "email":"user1@example.com",
   "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"
}

1.3. Change the attribute name of correlation id　相関IDの属性名を変更
設定オプションで相関IDの属性名を変更することができます。

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'})

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"
}

1.4. Change the log lervel ログのlervelを変更する

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'})

2. Event handlers イベントハンドラ
デコレータは、Kinesis、SNS、SQSイベントなどから配列を解析するなど、一般的なラムドバタスクをシンプルに実装します。

提供されているデコレータは以下の通りです。
2.1. common
commonデコレータは、例外に直面したときにイベント、レスポンス、エラー情報を出力することができます。

from jeffy.framework import get_app
app = get_app()

app.logger.update_context({
   'username': 'user1',
   'email': 'user1@example.com'
})

@app.handlers.common()
def handler(event, context):
    ...

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"
}

2.2. rest_api
API Gatewayイベントのデコレータです。リクエストヘッダから相関IDを自動的に取得し、レスポンスヘッダに相関IDを設定します。

デフォルトのエンコーディングは 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.'
        })
    }

デフォルトのヘッダ名は 'x-jeffy-correlation-id' です。この名前は設定オプションで変更できます。

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.'
        })
    }

2.3. sqs
sqsイベントのデコレータです。SQSのイベントソースから「event.Records」リストを項目ごとに自動でパースし、Lambdaのメイン処理内での扱いを容易にしています。

デフォルトのエンコーディングは 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
    """

2.4. sqs_raw
sqsの生イベント(全てのメタデータを含む)をデコレータで処理します。SQSのイベントソースから "event.Records "リストを自動で解析し、各レコードをLambdaのメイン処理に渡します。

デフォルトのエンコーディングは jeffy.encoding.json.JsonEncoding です。

from jeffy.framework import get_app
app = get_app()

@app.handlers.sqs_raw()
def handler(event, context):
    return event['body']

2.5. sns
SNSイベント用のデコレータです。SNSイベントソースのevent.Recordsリストを項目ごとに自動でパースし、Lambdaのメイン処理内での扱いを容易にしています。

デフォルトのエンコーディングは 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
    """

2.6. sns_raw
sqsの生イベント(すべてのメタデータを含む)のデコレータ。SNSのイベントソースから "event.Records "リストを自動で解析し、各レコードをLambdaのメイン処理に渡します。

デフォルトのエンコーディングは 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']

2.7. kinesis_streams
Amazon Kinesisのストリームイベントのデコレータです。Kinesisイベントソースのevent.Recordsリストを各項目ごとに自動解析し、Lambdaのメイン処理内で扱いやすいようにbase64でデコードします。

デフォルトのエンコーディングは 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,
     [
         ,
         
     ]

    event['foo'] value is 1 and 2, event['correlation_id'] is correlation_id you should pass to next event
    """

2.8. kinesis_streams_raw
sqsの生イベント(すべてのメタデータを含む)のデコレータ。Kinesis Data Streamsのイベントソースから "event.Records "リストを自動で解析し、各レコードをLambdaのメイン処理に渡します。

デフォルトのエンコーディングは 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']

2.9. dynamodb_streams
Dynamodbストリームイベント用のデコレータです。Dynamodbイベントソースのevent.Recordsリストを項目に自動でパースし、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
    """

2.10. s3
S3イベントのデコレータ。トリガされたS3オブジェクトとS3バケットとキー名からのボディストリームを自動的にLambdaにパースします。

このハンドラには s3:GetObject 権限が必要です。

デフォルトのエンコーディングは 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 matadata

2.11. schedule
スケジュールイベントのデコレータです。
メインのLambda処理の前に相関IDを取得するだけです。

from jeffy.framework import setup
app = setup()

@app.handlers.schedule()
def handler(event, context):
    ...

3. SDK
JeffyにはAWS SDK(boto3)のオリジナルラッパークライアントがあります。このクライアントは、イベントのペイロードにcorrelation_idを自動的に注入し、指定された(またはデフォルトの)エンコーディングにエンコードします。
3.1. Kinesis Clinent
デフォルトのエンコーディングは 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'
    )

3.2. SNS Client
デフォルトのエンコーディングは 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'
    )

3.3. SQS Client
デフォルトのエンコーディングは jeffy.encoding.json.JsonEncoding です。

Jeffyは'message'引数をトレースID属性を持つJSONデータにラップし、キューに送信します。

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'
    )

上記のサンプルでは、あなたの'hello world'は以下のJSONデータとしてキューに送信されます(デフォルトでは、jeffyはトレースID属性として'correlation_id'を付加します)。

{
    "message": "hello world",
    "correlation_id": ""
}

メッセージとしてdict値を使用することもできます。また、メッセージとしてdict型を使用することもできます。Jeffyはあなたのdict値を'message'属性に挿入します。

注意: デフォルトのエンコーダである JsonEncoder を使うことを推奨します。
JsonEncoderでなくても構いませんが、dict型をエンコードできるカスタムエンコーダを実装する必要があります
(現在のところ、Jeffyが提供しているBytesEncoderはこのユースケースには対応していません)。

3.4. S3 Client
デフォルトのエンコーディングは 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'
    )

4. Encoding
Jeffy encoderは、AWS Lambdaと他のAWSサービス間でメッセージの仕様を整合させるために型として機能します。これにより、マイクロサービスの開始から終了までのメッセージの型を定義することで、より安全で効率的かつ堅実にLambda機能を実装することができます。

各ハンドラやSDKクライアントはデフォルトのエンコーディングを持っており、Pythonオブジェクトからのデータを自動的にエンコード/デコードします。また、エンコーディングを変更することもできます。
現在、選択できるエンコーディングは以下の通りです。
・jeffy.encoding.bytes.BytesEncoding : バイナリデータを使用する場合。
・jeffy.encoding.json.JsonEncoding : JSON形式を使用。現時点でのJeffyではAmazon S3以外はこれをデフォルトのエンコーダとして使用しています。

Jeffyは内部的にはboto3のラッパーであるAWS SDK for pythonを利用しています。Lambda機能からAWS Servicesにメッセージを公開する場合、Jeffyはデフォルトのエンコーダーで自動的にエンコードします。一方、AWS Lambdaからメッセージを購読する場合は、イベントソースである各AWSサービスのエンコーダーで自動的にデコードします。後はLambda関数内でメッセージを扱うことができます。

また、各エンコーディングクラスには、「bytes」のデータを独自のエンコーディングに変換する「encode」メソッドがあります。このように、開発者は使用したいエンコーディングを設定することができます。

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'
    )

このエンコーディング機構は、Pub - Sub スタイルで構成されるイベント駆動型アーキテクチャのメッセージの一貫した仕様を簡素化します。
※Pub/Subメッセージングモデル
http://itdoc.hitachi.co.jp/manuals/link/cosmi_v0870/APKC/EU070377.HTM

5. Validation
5.1. JSONSchemaValidator
JsonSchemaValidator は、定義した以下の json スキーマでイベントのペイロードを自動的にバリデーションします。バリデーションに失敗した場合は ValidationError 例外を発生させます。

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.'
        })
    }

Requirements以下は省略。

-AWS, IT, Jeffy, Lambda, Python, プログラム
-, , , ,