2024.06.10

PythonでDIやってみた

tsucci

こんにちは、こんばんは、つっちーです。最近（執筆は24年5月）は生成AI関連が賑やかですね。AIといえばPythonということで今回もPythonネタでいこうと思います。

突然ですが、皆さんDIしてますか？

DI（Depencency Injection: 依存性注入）はかの有名なクリーンアーキテクチャで登場するアレですが、関心を分離することでテスタブルなコードを書こうというアレですが、実際やってみると依存解決が煩雑になりがちですよね（筆者だけかも知れませんが）。

各言語で色々なDIフレームワークが提供されているかと思いますが（Go言語だとwireとか）、PythonのInjectorを使ってみようというのが今回の趣旨です。

Contents

構成
Injector
1. 基本の依存解決
2. 抽象の依存解決
まとめ

構成

今回はLambda関数を実装する想定で、以下のようなディレクトリ構成を考えます。

.
├── domain
│   ├── user.py
│   └── user_repository.py
├── infra
│   ├── __init__.py
│   └── dynamodb.py
├── lambda_handler.py
└── usecase
    ├── __init__.py
    └── get_user.py

domain層

シンプルなユーザモデルを定義します。

# domain/user.py
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    age: int

# domain/user_repository.py
from typing import Protocol, runtime_checkable
from domain.user import User

@runtime_checkable
class UserRepository(Protocol):
    def get_item(self, user_id: int) -> User: ...

usecase層

Pythonにインターフェースという概念はないので、抽象クラス（ABC）を使うかtyping.Protocolを使うかで宗教が分かれるかと思いますが、今回は後者を採用します。

# usecase/get_user.py
from dataclasses import dataclass
from typing import Protocol

from botocore.exceptions import ClientError
from domain.user import User
from domain.user_repository import UserRepository
from pydantic import BaseModel

class GetUserRequest(BaseModel):
    user_id: int


class GetUserResponse(BaseModel):
    user: User

class IGetuser(Protocol):
    def execute(self, req: GetUserRequest) -> GetUserResponse: ...

class GetUser:
    repository: UserRepository

    def execute(self, req: GetUserRequest) -> GetUserResponse:
        try:
            user = self.repository.get_item(req.user_id)
            return GetUserResponse(user=user)
        except ClientError as e:
            raise e

infra層

外部I/Fとしてinfra層を定義し、今回はDynamoDBを使います。

# infra/dynamodb.py
from dataclasses import dataclass

from boto3 import Session
from boto3.dynamodb.types import TypeDeserializer, TypeSerializer
from botocore.exceptions import ClientError
from domain.user import User
from mypy_boto3_dynamodb import DynamoDBClient

@dataclass
class DynamoDB:
    client: DynamoDBClient
    table_name: str

    def __init__(self, session: Session, table_name: str):
        self.client = session.client("dynamodb")
        self.table_name = table_name

    def get_item(self, user_id: int) -> User:
        try:
            res = self.client.get_item(
                TableName=self.table_name,
                Key=TypeSerializer().serialize({"id": user_id})["M"],
            )
            item = {
                k: TypeDeserializer().deserialize(v) for k, v in res["Item"].items()
            }
            return User.model_validate(item)
        except ClientError as e:
            raise e

handler

Lambdaハンドラーを定義します。

# lambda_handler.py
import json
import os

from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.typing import LambdaContext
from boto3 import Session
from botocore.exceptions import ClientError
from infra.dynamodb import DynamoDB
from pydantic import BaseModel
from usecase.get_user import GetUser, GetUserRequest

logger = Logger()

class LambdaEventRequest(BaseModel):
    id: int

class LambdaEventResponse(BaseModel):
    status_code: int
    body: str


@logger.inject_lambda_context(log_event=True)
def lambda_handler(event: dict, context: LambdaContext) -> LambdaEventResponse:
    body = json.loads(event["body"])
    lambda_event = LambdaEventRequest.model_validate(body)

    try:
        # ここが煩雑
        dynamodb = DynamoDB(
            session=Session(profile_name=os.getenv("AWS_PROFILE")),
            table_name=os.getenv("DYNAMODB_TABLE_NAME", ""),
        )
        usecase = GetUser(repository=dynamodb)
        req = GetUserRequest(user_id=lambda_event.id)
        res = usecase.execute(req)
    except ClientError as e:
        logger.exception(e)
        return LambdaEventResponse(
            status_code=500,
            body="Internal Server Error",
        )

    return LambdaEventResponse(
        status_code=200,
        body=repr(res.user),
    )

以下のコードがやや煩雑な印象を受けます。せっかく関心を分離しているのにhandlerがinfra層に依存するような構図になってしまっています。別にDIコンテナを定義してそこで依存解決を行うのがよくあるパターンかと思います。

dynamodb = DynamoDB(
    session=Session(profile_name=os.getenv("AWS_PROFILE")),
    table_name=os.getenv("DYNAMODB_TABLE_NAME", ""),
)
usecase = GetUser(repository=dynamodb)

Injector

https://github.com/python-injector/injector

前述の通りPythonのDIフレームワークの一つです。

基本の依存解決

最もシンプルな使い方です。注入したいオブジェクトに@injectデコレータを設定し、注入する対象に@providerデコレータを付与します。

# infra/dynamodb.py
@inject # 付与
@dataclass
class DynamoDB:
    client: DynamoDBClient
    table_name: str

    def __init__(self, session: Session, table_name: str):
        self.client = session.client("dynamodb")
        self.table_name = table_name

実際の依存性注入にはModuleという概念を使います。今回はinfra/__init__.pyに記述します。

# infra/__init__.py
from injector import Module, provider

class DynamoDBModule(Module):
    @provider # 付与
    def session(self) -> Session:
        return Session(profile_name=os.getenv("AWS_PROFILE_NAME"))

    @provider # 付与
    def table_name(self) -> str:
        return os.getenv("DYNAMODB_TABLE_NAME", "")

抽象の依存解決

usecaseは具象型（インターフェース）に依存しているので、依存解決には具象型が必要になりますがBinder.bind()を使うことで実装できます。こちらもusecase/__init__.pyに記述します。

# usecase/__init__.py
from domain.user_repository import UserRepository
from infra import DynamoDBModule
from infra.dynamodb import DynamoDB
from injector import Binder, Injector, Module
from usecase.get_user import GetUser

@dataclass
class GetUserModule(Module):
    injector: Injector

    def __init__(self) -> None:
　　　　　# ここで依存解決
        self.injector = Injector(
            [DynamoDBModule(), self.__class__.configure],
        )

    @classmethod
    def configure(cls, binder: Binder) -> None:
        # 抽象型に具象型を割り当てる（バインドする）
        binder.bind(UserRepository, to=DynamoDB)  # type: ignore

    def resolve(self) -> GetUser:
　　　　　# 依存解決したインスタンスを返す
        return self.injector.get(GetUser)

注）Mypyでtype-abstractエラーが出てしまうので今回は# type: ignoreで除外しています。どうやらMypyがProtocolに対応していないようです。

このクラスがDIコンテナとして機能します。Injector.get()というメソッドが、最終的に依存解決したusecaseのインスタンスを返します。

無事に依存解決ができました。あとはハンドラから呼び出すだけです。handlerがusecase/infraに関心を持っておらず分離できているのがいい感じです。

from usecase import GetUserModule

@logger.inject_lambda_context(log_event=True)
def lambda_handler(event: dict, context: LambdaContext) -> LambdaEventResponse:
    body = json.loads(event["body"])
    lambda_event = LambdaEventRequest.model_validate(body)

    req = GetUserRequest(user_id=lambda_event.id)

    try:
        # usecase/infraに関心を持たない
        usecase = GetUserModule().resolve()
        res = usecase.execute(req)
    ...