PythonでDIを行う(FastAPI, Djangoでのやり方も)
PythonのDIライブラリ
主に以下の2つがある。
- injector
- python-dependency-injector
DIとは
DI自体は以下の説明がわかりやすいです。
嬉しいことの一つとしてテストが書きやすくなる。
injector
を導入することで、外部からのパラメータの注入、外部操作するインスタンス生成を Module クラスなどに集めることができる。
「 Module クラスのパラメータをMockすれば単体で動かせるぞ」という状況を作りやすくなり、テストが書きやすくなることが期待できる。
DIは結合度(Coupling)
を減らし、凝集度(Cohesion)
を高めるのに役立つ原理。
- 結合度 (Coupling)
- 凝集度(Cohesion)
結合度と凝集度は、部品がどれだけ強く結ばれるか
ということを表している。
高結合
結合度が高い場合は、瞬間接着剤や溶接を使用するようなもの。
分解する簡単な方法は無い。
高凝集
凝集度が高い場合とは、ネジを使用するようなもの。
分解して組み立て直す、または別の方法で組み立てるのは非常に簡単。
これは、高結合の反対になる。
凝集度が高い = 結合度は低い
低結合は柔軟性をもたらし、コードの変更とテストが簡単になる。
Injector
# conding: utf-8
from abc import ABCMeta, abstractmethod
from injector import Injector, inject, Module
class abstractTire(metaclass=ABCMeta):
@abstractmethod
def rotate(self, angle: int) -> int:
return angle
class Tire(abstractTire):
def rotate(self, angle: int) -> int:
return angle
class Car:
@inject
def __init__(self, t: abstractTire):
if not isinstance(t, abstractTire):
raise Exception("t is not abstractTire.")
self.t = t
def drive(self, angle: int) -> int:
return self.t.rotate(self.h.rotate(angle))
class carDIModule(Module):
def configure(self, binder):
binder.bind(abstractTire, to=Tire)
if __name__ == '__main__':
injector = Injector([carDIModule()])
c = injector.get(Car)
print(c.drive(angle=10))
injector で利用するアノテーション
デコレータとして付与する
- @inject
- @singleton
- @provider
@inject
オブジェクト注入する際に宣言する。サービス層などで使う場合が多い。
型宣言が必須。
@singleton
インジェクトの度に新しいインスタンスを生成したくない時に宣言する。
リポジトリ 層のメソッドやサービス層のクラスに付与することが多い。
@provider
メソッドと戻り値をインジェクトに利用したい場合に宣言する。
シンプルなDI
from injector import singleton
@singleton
class TodoUseCase:
def register(self, todo: Todo) - > None:
print("call todo_usecase.register")
from injector import inject, singleton
class Todo:
""" """
@singleton
class TodoController:
@inject
def __init__(self, todo_usecase: TodoUseCase) -> None:
self.todo_usecase = todo_usecase
def create_todo(self, todo: Todo) -> None:
self.todo_usecase.register(todo)
if __name__=='__main__':
injector = Injector()
todo_controller: TodoController = injector.get(TodoController)
todo: Todo = Todo()
todo_controller.create_todo(todo)
メソッドに@inject
のデコレータを付与するだけで、勝手にDIをしてくれます。
これの場合、TodoController
クラスのデフォルトコンストラクタの引数であるtodo_usecase
には、自動的にTodoUseCase
のインスタンスが入ります。
DIコンテナを使用する
from injector import singleton
class AbstractTodoUseCase:
@abstractmethod
def register(self, todo: Todo) -> None:
raise NotImplementedError
@singleton
class TodoUseCase(AbstractTodoUseCase):
def register(self, todo: Todo) - > None:
print("call todo_usecase.register")
from injector import inject, singleton, Module
from .todo_usecase import AbstractTodoUseCase, TodoUseCase
class Todo:
""" """
@singleton
class TodoController:
@inject
def __init__(self, todo_usecase: AbstractTodoUseCase) -> None:
self.todo_usecase = todo_usecase
def create_todo(self, todo: Todo) -> None:
self.todo_usecase.register(todo)
class TodoDIModule(Module):
def configure(self, binder):
binder.bind(AbstractTodoUseCase, to=TodoUseCase, scope=singleton)
if __name__=='__main__':
injector = Injector([TodoDIModule()])
todo_controller: TodoController = injector.get(TodoController)
todo: Todo = Todo()
todo_controller.create_todo(todo)
TodoDIModule
でDIコンテナの設定ができる。
binder.bind
で抽象クラスと実クラスの結び付けができる。
引数に注入されるのをクラス以外にもする
TodoDIModule
では抽象クラスと実クラスを結びつけることで、実クラスが自動的に注入されるようになっているが、数値や文字列を自動的に注入されるようにしたい場合もある。
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__
で必要な引数を主に返す。
詳細は以下に詳しく載っている。
昔、自分が書いた記事がリンクとして貼られていた。(ありがとうございます! )
こちらの記事ではInstanceProvider
を用いれば、似たようなことができる模様。
python-dependency-injector
Dependency Injector
はオブジェクトの組み立てを支援するコンテナーとプロバイダーを提供する。
$ pip install dependency-injector
オブジェクトが必要な場合は、関数の引数のデフォルト値としてProvideマーカー
を配置する。(ここではmain関数の引数)
この関数を呼び出すと、フレームワークが依存関係を組み立てて注入する。
from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide
class Container(containers.DeclarativeContainer):
config = providers.Configuration()
api_client = providers.Singleton(
ApiClient,
api_key=config.api_key,
timeout=config.timeout.as_int(),
)
service = providers.Factory(
Service,
api_client=api_client,
)
@inject
def main(service: Service = Provide[Container.service]):
...
if __name__ == '__main__':
container = Container()
container.config.api_key.from_env('API_KEY')
container.config.timeout.from_env('TIMEOUT')
# コンテナを注入する対象のモジュールを指定する
container.wire(modules=[sys.modules[__name__]])
main() # <-- 依存性が自動的に注入される
with container.api_client.override(mock.Mock()):
main() # <-- モックに置き換えられた依存性が自動的に注入される
main()
関数を呼び出すと、Service
の依存関係が自動的に組み立てられ注入される。
テストを行うときは、container.api_client.override()
を呼び出して、実際のAPIクライアントをモックに置き換える。
main()を呼び出すと、モックが注入される。
任意のプロバイダーを別のプロバイダーで置き換えることができる。
他にもさまざまな環境向けにプロジェクトを環境設定するのにも役立つ。
今回の例では、APIクライアントを開発環境もしくはステージング環境のスタブに置き換えている。
オブジェクトの組み立てはコンテナに統合される。依存性の注入が明示的に定義
される。
これにより、アプリケーションの動作を理解し、変更することが容易
になる。
以下に詳しく書かれている。
下記のリポジトリはFastAPIを使用した上記の方が作ったサンプル
python-dependency-injectorを使用したFastAPIのサンプル
$ tree
./
├── giphynavigator/
│ ├── __init__.py
│ ├── application.py
│ ├── containers.py
│ ├── endpoints.py
│ ├── giphy.py
│ ├── services.py
│ └── tests.py
├── config.yml
└── requirements.txt
container.py
DIコンテナを定義するファイル
endpoints.py
ルーティングのファイル
application.py
エントリポイントを定義している。
FastAPI + SQLAlchemyとの組み合わせ
Clean Architectureへの応用
参考例
DIを使う機会はテストを作成しやすくするために使う以外にも、整理されたコードを書く目的としてもクリーンアーキテクチャで使う機会が多い。
型付け
こちらでも出てきますが、型付けというのがDIでは重要になります。依存性逆転の法則(Dependency Inversion Principle)
というものです。
動的型付け言語ではあるものの、Pythonにもtypingという機能で型付けの機能があります。(実行時は違う型でも動いてしますので、pyrightやmypyなどが必要)
インタフェースがない場合の方法
DIをするには、インターフェースの機能が必要ですが、Pythonにはインターフェースがありません。
その代わりに抽象クラスがあるため、抽象クラスを使用します。
抽象クラスは実装を持ててしまいますが、@abstractmethod
というデコレータを付与することで、実装を持たせないようにします。
以下はサンプルです。
import asyncio
from abc import ABC, ABCMeta, abstractmethod
class InterfaceSampleRepository(metaclass=ABCMeta):
@abstractmethod
async def get(self, resource_id: str) -> dict:
raise NotImplementedError
class SampleRepository(InterfaceSampleRepository):
async def get(self, resource_id: str) -> dict:
return {"id": id}
class SampleApplicationService:
repository: InterfaceSampleRepository
def __init__(self, repository: InterfaceSampleRepository):
self.repository = repository
def get(self, resource_id: str) -> dict:
return asyncio.run(self.repository.get(resource_id))
SampleApplicationService(repository=SampleRepository()).get()
SampleApplicationService
が知っているRepository
は抽象クラスであるInterfaceSampleRepository
です。
そのため、SampleApplicationService
はコンストラクタの引数であるrepositoryの実装の中身を知りません。
ただし、こちらの方法ではDIコンテナを使用してない方法です。
ユニットテストでモックを用いる
冒頭でも述べたようにテストのしやすさもあります。以下のコードを
import unittest
from unittest.mock import AsyncMock
from sample import SampleApplicationService
class MockSampleRepository(InterfaceSampleRepository):
async def get(self, resource_id: str) -> dict:
raise NotImplementedError
class TestSampleUsecase(TestCase):
def test_get(self):
get_mock = AsyncMock(return_value={"id": "0002"})
repository = MockSampleRepository()
repository.get = get_mock
usecase = SampleApplicationService(repository=repository)
self.assertEqual(usecase.get("0002"), {"id": "0002"})
get_mock.assert_called_with("0002")
MagicMock/AsyncMockが提供されており、これを使えばメソッドの差し替えが行え、呼び出されたことを確認するverifyも実行できる。
テスト時にはDI
を活用して、Usecase層にテスト用となるダミーのRepository
を差し込み、テストするレイヤーのみの動作検証を行うようにする。
また上記のMagicMock/AsyncMock
のようなダミークラスを用意せず、実際に使われるクラスインスタンスを注入した上で、差し替えを行っても同様のことが実現できる。
Injectorを用いた実例
django-ninja-extraを用いた応用