京セラコミュニケーションシステム株式会社技術開発センター ICT技術開発部先端技術開発課の今村です。
私はJavaやC++言語を使用した開発が多かったのですが、最近Python言語で開発する機会が増えてきました。
JavaのときはSpringFrameworkなどで比較的簡単に外部設定を用いたDIが使えていたのですが、Pythonでは手探りで実装したので、その内容を残しておきます。

はじめに

データを読み込み・加工・別なフォーマットで出力という簡易的ETL機能を開発しました。
その中で、以下仕様の実現を目指しDIを活用することにしました。
１）読み込み元のデータ種類は複数存在するので、データ種類毎にクラスを用意し、
実行時にユーザーが指定したクラスを使用する
２）データ読み込みクラスの数は増える可能性があるので、クラスをオンコーディングしない。

対象読者

Python言語で開発した経験のある人で、DIをしたいと考えているエンジニア

前提条件

今回の記事にのせているコードは、以下の環境で動作確認しています。

	バージョン
Python	3.12.4
Injector	0.22.0

DIフレームワークパッケージの追加

今回は社内利用のみのためInjectorパッケージを使用しました。
BSD-3ライセンスですので、商用利用される場合はご注意ください。

下記コマンドにてInjectorパッケージをインストールします。

pip install injector

サンプルプログラムについて

実案件は少し複雑な構成になっているので、本記事では、インスタンスを注入してメソッドを実行し、文字列を表示するというシンプルなプログラムとしました。

Injectionを使用しない場合の実行部分コードは以下になります。

from extractor import ExtractorB
from executor import Executor


if __name__ == '__main__':
    extractor = ExtractorB()
    extractor.name = 'extractor.ExtractorB'
    executor = Executor(extractor)
    executor.run()

データ読み込み用のExtractorをExecutorにセットして、Executorのrunを呼び出すことで、読み込みクラスを実行しメッセージを表示するという簡単なものです。

ここで、データ読み込みの種類が増えた場合、6行目のExtractorを差し替える必要がありますが、条件分岐などで入れ替えるのは煩雑なコーディングになってきます。

使用するのが1ヵ所であれば、動的クラス生成を行うことでも代用できるかもしれませんが、複数個所ある場合はそうもいきません。そこでDIを使用することにしました。

DI サンプルプログラム

上記のプログラムをDIを使用したコードに変えてみましょう。

注入されるクラス

まずは実際の処理（今回の場合はメッセージを表示する）を行うクラスの定義になります。

複数のデータ種類（読み込みクラス）に対応するため、インターフェースを定義します。

iextractor.py

from abc import ABCMeta, abstractmethod


class IExtractor(metaclass=ABCMeta):

    def __init__(self) -> None:
        pass

    @abstractmethod
    def setName(self, name: str) -> None:
        pass

    @abstractmethod
    def run(self) -> None:
        pass

次に実体のデータ読み込みクラスです。今回は、AとBの2つにしました。runメソッドが処理実行のイメージです。
基底クラスを作っては？というご意見をいただきそうですが、まるコピーにしときます。

extractor.py

from iextractor import IExtractor


class ExtractorA(IExtractor):

    def __init__(self) -> None:
        pass

    def setName(self, name: str) -> None:
        self.name = name

    def run(self) -> None:
        # 実際はデータ読み込みするなどの処理
        print(f'Extractor A:{self.name}')


class ExtractorB(IExtractor):

    def __init__(self) -> None:
        pass

    def setName(self, name: str) -> None:
        self.name = name

    def run(self) -> None:
        # 実際はデータ読み込みするなどの処理
        print(f'Extractor B:{self.name}')

依存性を持ったクラス

処理を行うクラスに依存している、データ読み込みを実行するクラスを定義します。
runメソッドでデータ読み込み処理を実行します。

@injectデコレータが依存性注入を行う宣言になります。IExtractorをinjectの対象とするイメージです。

executor.py

from injector import inject
from iextractor import IExtractor


class Executor():

    @inject
    def __init__(self, extractor: IExtractor) -> None:
        self.__extractor = extractor

    def run(self) -> None:
        self.__extractor.run()

注入するクラスを提供するクラス

IExtractorインターフェース（クラス）に実体インスタンスを提供するためのプロバイダーを定義します。

実体クラス定義のimportを書き足していかなくてもいいように、クラス名の文字列からgetattrで動的にクラスを取得しインスタンス生成するようにしました。
Binderでもできると思いますが、指定にもとづいてインスタンス生成までに他の前処理などもできることを考慮して、ここで動的生成するようにしています。

injectorがproviderに受け渡すための情報をデータクラスとして持ちます。

configuration.py

from dataclasses import dataclass


@dataclass
class Configuration():

    extractor_name: str = ''

次に実体インスタンスを提供するクラスです。@providerデコレータがクラス提供を行うメソッドであることの宣言になります。

@injectデコレータのスコープ中（今回は__init__メソッド）に出てくるクラスと@providerデコレータを指定したメソッドの戻り値クラスが同一の場合に、そのメソッドを呼び出し生成されたインスタンスが注入されます。
今回の場合は、provide_extractorメソッドの戻り値IExtractorが、Executorの__init__の引数IExtractorと一致するので、これが注入の対象となります。

※get_classメソッドはインスタンスメソッドでなくてもよいです。

executor_module.py

import importlib
from injector import Module, provider, singleton
from configuration import Configuration
from iextractor import IExtractor
from exception import ClassNotFoundError


class ExecutorModule(Module):

    @singleton
    @provider
    def provide_extractor(self, configuration: Configuration) -> IExtractor:
        cls: IExtractor = self.get_class(configuration.extractor_name)
        cls.setName(configuration.extractor_name)
        return cls

    def get_class(self, class_name: str) -> any:
        module_name, class_name = class_name.rsplit(".", 1)
        module = importlib.import_module(module_name)
        if not hasattr(module, class_name):
            raise ClassNotFoundError(module_name, class_name)
        cls = getattr(module, class_name)
        return cls()

本体の処理には関係しませんが、get_classメソッドの中で、独自例外の定義をし標準エラーと区別できるようにしているので、そちらのクラスも掲載しておきます。

exeption.py

class ClassNotFoundError(Exception):

    def __init__(self, module_name: str = '', class_name: str = ''):
        super().__init__(self)
        self.__module_name = module_name
        self.__class_name = class_name

    def __str__(self) -> str:
        return (
            f'{self.__module_name}パッケージの{self.__class_name}が存在しません。'
        )

実行

最後にExecutorの実行部分を定義します。

configureでproviderの引数に渡すConfigureを生成しています。ここでは、Configureのextractor_nameに注入したいクラスのクラス名を設定し、bindでConfigurationクラスに紐付けています。

InjectorにBinderとModule(provider)を渡すことで、DI構成を定義しています。
Injectorのgetメソッドに@injectデコレータのあるクラスを指定することで、DIした結果のクラスインスタンスを取得することできます。

あとは、そのインスタンスのメソッドを呼び出して実行するだけです。

main.py

from injector import Injector, Binder, singleton
from executor_module import ExecutorModule
from configuration import Configuration
from executor import Executor


def configure(binder: Binder):
    configuration = Configuration(extractor_name='extractor.ExtractorB')
    binder.bind(Configuration, to=configuration, scope=singleton)


if __name__ == '__main__':
    di = Injector([configure, ExecutorModule()])
    executor: Executor = di.get(Executor)
    executor.run()

実行

python main.py

実行結果

Extractor B:extractor.ExtractorB

8行目のextractor_nameに指定するクラス名を変えることで、データ読み込みクラスを変えて実行することができます。

また、本サンプルでは、configureで'extractor.ExtractorB'とソースに直接記載していますが、外部のxmlやjsonファイルまたはデータベースに定義し、その内容を取得・設定することでJavaのSpringFrameworkのようにソースを修正することなく、機能に合わせたクラスの組み合わせで実行できる独自の処理フレームワークを作ることができます。

所属部署について

隼人事業所

隼人事業所は、工場の中に事務所があるという利点を活かし、ものづくりの現場に貢献できるシステム開発に取り組んでいます。
IT技術を活用して、新しい価値を創出をしていくことを目指しています。

おわりに

このサンプルでは、あまりDIの嬉しさがわからないかもしれませんが、テストやフレームワーク作成などを行う場合に非常に便利なデザインモデルです。Pythonでも先人のかたがDIパッケージを作ってくださっていて非常に助かりました。

これからInjectorを使用してDIやってみようという人の参考になれば幸いです。