Zenn
Open11

PythonでDIを行う(FastAPI, Djangoでのやり方も)

shimakaze_softshimakaze_soft

DIとは

DI自体は以下の説明がわかりやすいです。

https://qiita.com/mkgask/items/d984f7f4d94cc39d8e3c#依存とは

嬉しいことの一つとしてテストが書きやすくなる。
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))
shimakaze_softshimakaze_soft

injector で利用するアノテーション

デコレータとして付与する

  • @inject
  • @singleton
  • @provider

@inject

オブジェクト注入する際に宣言する。サービス層などで使う場合が多い。
型宣言が必須。

@singleton

インジェクトの度に新しいインスタンスを生成したくない時に宣言する。
リポジトリ 層のメソッドやサービス層のクラスに付与することが多い。

@provider

メソッドと戻り値をインジェクトに利用したい場合に宣言する。

shimakaze_softshimakaze_soft

シンプルなDI

todo_usecase.py
from injector import singleton

@singleton
class TodoUseCase:
    def register(self, todo: Todo) - > None:
        print("call todo_usecase.register")
todo_controller.py
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のインスタンスが入ります。

shimakaze_softshimakaze_soft

DIコンテナを使用する

todo_usecase.py
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")
todo_controller.py
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では抽象クラスと実クラスを結びつけることで、実クラスが自動的に注入されるようになっているが、数値や文字列を自動的に注入されるようにしたい場合もある。

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__ で必要な引数を主に返す。

詳細は以下に詳しく載っている。

https://dev.classmethod.jp/articles/python-injector-launch01/#toc-10

昔、自分が書いた記事がリンクとして貼られていた。（ありがとうございます！ ）
https://speakerdeck.com/shimakaze_soft/python-de-dependency-injection-di-woyaruniha

こちらの記事ではInstanceProviderを用いれば、似たようなことができる模様。

https://zenn.dev/515hikaru/articles/python-injector-module

shimakaze_softshimakaze_soft

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クライアントを開発環境もしくはステージング環境のスタブに置き換えている。

オブジェクトの組み立てはコンテナに統合される。依存性の注入が明示的に定義される。
これにより、アプリケーションの動作を理解し、変更することが容易になる。

以下に詳しく書かれている。

https://zenn.dev/shimat/articles/4be773f427c502#課題2.-"singletonのdi"がやりづらい

下記のリポジトリはFastAPIを使用した上記の方が作ったサンプル
https://github.com/shimat/fastapi_di_sample/tree/8464f851101d11bd6b4832cb61282a829c2cd8ed/src/fastapi_di_sample

shimakaze_softshimakaze_soft

python-dependency-injectorを使用したFastAPIのサンプル

$ tree
./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── endpoints.py
│   ├── giphy.py
│   ├── services.py
│   └── tests.py
├── config.yml
└── requirements.txt

https://python-dependency-injector.ets-labs.org/examples/fastapi.html

container.py

DIコンテナを定義するファイル

endpoints.py

ルーティングのファイル

application.py

エントリポイントを定義している。

FastAPI + SQLAlchemyとの組み合わせ

https://python-dependency-injector.ets-labs.org/examples/fastapi-sqlalchemy.html

shimakaze_softshimakaze_soft

Clean Architectureへの応用

参考例

https://github.com/t-tiger/Python-CleanArchitecture-Example

DIを使う機会はテストを作成しやすくするために使う以外にも、整理されたコードを書く目的としてもクリーンアーキテクチャで使う機会が多い。

型付け

こちらでも出てきますが、型付けというのがDIでは重要になります。依存性逆転の法則(Dependency Inversion Principle)というものです。
動的型付け言語ではあるものの、Pythonにもtypingという機能で型付けの機能があります。（実行時は違う型でも動いてしますので、pyrightやmypyなどが必要）

インタフェースがない場合の方法

DIをするには、インターフェースの機能が必要ですが、Pythonにはインターフェースがありません。
その代わりに抽象クラスがあるため、抽象クラスを使用します。

抽象クラスは実装を持ててしまいますが、@abstractmethodというデコレータを付与することで、実装を持たせないようにします。

以下はサンプルです。

sample.py
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コンテナを使用してない方法です。

shimakaze_softshimakaze_soft

ユニットテストでモックを用いる

冒頭でも述べたようにテストのしやすさもあります。以下のコードを

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のようなダミークラスを用意せず、実際に使われるクラスインスタンスを注入した上で、差し替えを行っても同様のことが実現できる。