はじめに
こんにちは、筧( @TakaakiKakei )です。
所属しているチームでは、開発言語として python をよく使っています。 そして最近、コード品質向上のために injector を導入しました。
alecthomas/injector: Python dependency injection framework, inspired by Guice
injector は python で DI を実現するフレームワークです。 私は DI 初心者で、同僚のコードを読み解きながら理解を進めている状況です。 そこで今回は学んだことを、本ブログにアウトプットします。
何が嬉しいのか
嬉しいことの一つとしてテストが書きやすくなることが挙げられます。 injector を導入することで、外部からのパラメータの注入や、外部操作するインスタンス生成を Module クラスなどに集めることができます。 これによって、「 Module クラスのパラメータをMockすれば単体で動かせるぞ」という状況を作りやすくなり、テストが書きやすくなることが期待できます。
前提
所属チームで採用している階層構造について
従来
injector 採用前の階層構造です。
.
└── src
├── __init__.py
├── handlers
│ ├── __init__.py
│ ├── hello.py
├── services
│ ├── __init__.py
│ └── slack.py
└── use_cases
├── __init__.py
└── hello.py最新
injector 採用後の階層構造です。 modules 層を追加しました。
.
└── src
├── __init__.py
├── handlers
│ ├── __init__.py
│ └── hello.py
├── modules
│ ├── __init__.py
│ └── hello.py
├── services
│ ├── __init__.py
│ └── slack.py
└── use_cases
├── __init__.py
└── hello.py各階層の役割
以下が各階層の役割です。
- handlers
- プログラムのエントリーポイントを定義
- modules
- 外部からのパラメータの注入や、外部操作するインスタンス生成を定義
- services
- slack などのサービスを定義
- use_cases
- ビジネスロジックを定義
なお、従来の階層構造では、modules 層の役割を、use_cases 層が担っていました。
DI(Dependency Injection) とは
解釈の一つとして、「オブジェクト注入」と言えます。 依存元クラスが具象クラスに依存しているのを、抽象クラスに依存させることで、コンポーネント間の依存関係を削除します。 これによって、コードの変更をしやすくなったり、前述のようにテストが書きやすくなるソフトウェアパターンです。 以下の資料が読み物としてわかりやすかったので、リンクを置いておきます。
Python で Dependency Injection(DI) をやるには？ - Speaker Deck
injector で利用するアノテーションについて
コード紹介で出てくる、injector のアノテーションについて説明します。
@inject- オブジェクト注入する際に宣言します。
- 所属チームでは、modles/ services/ use_cases の層の
__init__に付与することが多いです。 - 型宣言が必須です。
@singleton- インジェクトの度に新しいインスタンスを生成したくない時に宣言します。
- 所属チームでは、modles 層のメソッドと services/ use_cases 層のクラスに付与することが多いです。
@provider- メソッドと戻り値をインジェクトに利用したい場合に宣言します。
- 所属チームでは、modles 層のメソッドに付与することが多いです。
コード例
前提
event 内の title と text の内容を Slack の Webhook 経由で Slack 通知するだけのサービスです。 コード内のアノテーションの意味は前述を参照ください。 また以下のライブラリは poetry などでインストール前提です。
- python
- requests
- boto3
- injector
- aws-lambda-powertools
handlers 層
src/handlers/notify_message.py
from typing import Dict
from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.utilities.typing import LambdaContext
from injector import Injector
from src.modules.notify_message import NotifyMessageModule
from src.use_cases.notify_message_to_slack import NotifyMessageToSlackUseCase
logger = Logger()
tracer = Tracer()
# Injector()内で、利用する Module を宣言
# Module 内では、外部からのパラメータの注入や、外部操作するインスタンス生成を定義
di = Injector([NotifyMessageModule()])
@tracer.capture_lambda_handler
@logger.inject_lambda_context(log_event=True)
def handler(event: Dict, context: LambdaContext) -> Dict:
"""
notify_message のエントリーポイント
event:
{
"title": "TEST",
"text": "hello"
}
"""
try:
# di.get()内で、利用する UseCase を宣言し、各インスタンスを生成
notify_message_use_case = di.get(NotifyMessageToSlackUseCase)
# UseCaseのインスタンスメソッドの exec を実行
notify_message_use_case.exec(event)
return event
except Exception as e:
logger.exception("throw exception in notify_message")
raise e- プログラムのエントリーポイントを定義しています。ハイライトの箇所が DI に関する主な処理です。
- 上記コード内のアノテーションは AWS Lambda Powertools 関連なので無視いただいても結構です。
- handler 外に
di = Injector(...)を書くことで、singleton ルールで書いてあるものが global 領域のメモリ保持に保持されます（ウォームスタート）。キャッシュされるので、SecretManager の値を更新したのにすぐに反映されないといった事象があるので注意しましょう。もし handler 内に書いた場合は、キャッシュされず、singleton ルールで書いてあるものが lambda に呼ばれる度に作られます。
modules 層
src/modules/notify_message.py
import os
import requests
from injector import Module, provider, singleton
from src.services.slack import SlackSession, SlackWebhookUrl
class NotifyMessageModule(Module):
@singleton
@provider
# src.services.slack の SlackService クラスの __init__ で必要な引数を返す
def slack_webhook_url(self) -> SlackWebhookUrl:
return os.environ["SLACK_WEBHOOK_URL"]
@singleton
@provider
# src.services.slack の SlackService クラスの __init__ で必要な引数を返す
def slack_session(self) -> SlackSession:
return requests.Session()- 外部からのパラメータの注入や、外部操作するインスタンス生成を定義しています。
- 利用する Service クラスの
__init__で必要な引数を主に返します。
services 層
src/services/slack.py
import json
from typing import Dict, NewType
import requests
from aws_lambda_powertools import Logger
from injector import inject, singleton
# 型エイリアス
SlackWebhookUrl = NewType("SlackWebhookUrl", str)
SlackSession = NewType("SlackSession", requests.Session)
logger = Logger(child=True)
@singleton
class SlackService:
@inject
def __init__(self, session: SlackSession, webhook_url: SlackWebhookUrl):
self._session = session
self._webhook_url = webhook_url
def send_msg(self, payload: Dict) -> None:
data = json.dumps(payload)
resp = self._session.post(self._webhook_url, data=data)
logger.info(f"resp.text: {resp.text}")
logger.info(f"payload: {data}")- 外部サービスなどを定義する層で、slack のサービスを定義しています。
- ハイライトの箇所を参照すると、NotifyMessageModule で定義していたものがあることが分かります。
use_cases 層
src/use_cases/notify_message_to_slack.py
from typing import Dict
from aws_lambda_powertools import Logger
from injector import inject, singleton
from src.services.slack import SlackService
logger = Logger(child=True)
@singleton
class NotifyMessageToSlackUseCase:
@inject
def __init__(self, slack_service: SlackService):
self.slack_service = slack_service
def exec(self, event: Dict) -> None:
blocks = [
{
"type": "header",
"text": {"type": "plain_text", "text": event["title"]},
},
{
"type": "section",
"text": {"type": "mrkdwn", "text": event["text"]},
},
]
payload = {"blocks": blocks}
self.slack_service.send_msg(payload)
return- ビジネスロジックを定義しています。payload で送信内容を定義して、インスタンス作成済みの SlackService の send_msg メソッドを使って、Slack チャンネルにメッセージ送信しています。
- NotifyMessageToSlackUseCase の 初期値は、
src/handlers/notify_message.pyの31行目の di.get() 時に注入しています。 - NotifyMessageToSlackUseCase の exec は、
src/handlers/notify_message.pyの33行目で実行されます。
おわりに
最後まで読んでいただきありがとうございます。
テストコードは、また別途紹介できたらなと思います。 本ブログが皆さんの役に立っていると嬉しいです。
それではまた！
参考
- Python の DI コンテナ実装の紹介と活用例 - All You Need Is Writing
- 【Python】injectorでDIコンテナを実装する - Qiita
- 依存性の注入とinjector
- Google Guice 使い方メモ - Qiita
- PythonでDI(Dependency Injection) - Qiita
更新履歴
- 2022/11/22
di = Injector(...)を handler 内外に書くことでどのように変わるか追記。- コード例では、
di = Injector(...)を handler 外で書くように変更
17
