0_Python_DI_Sample_with_Protocol.md

Protocolの記事用スクリプト

main.py

from typing import Protocol, Optional, Union
from dataclasses import dataclass
from dependency_injector import containers, providers
import pytest
from unittest.mock import Mock

# --- Data Models ---

@dataclass
class UserDetail:
    name: str
    email: str

@dataclass
class User:
    id: int
    detail: UserDetail

# --- Protocols (Interfaces) ---

class UserRepositoryProtocol(Protocol):
    def fetch(self, user_id: int) -> Optional[User]:
        ...

    def put(self, data: Union[User, UserDetail]) -> User:
        ...

# --- Concrete Implementations ---

class UserRepositoryOnMemory:
    def __init__(self):
        self._users: dict[int, User] = {}
        self._next_id = 1

    def fetch(self, user_id: int) -> Optional[User]:
        return self._users.get(user_id)

    def put(self, data: Union[User, UserDetail]) -> User:
        if isinstance(data, UserDetail):
            new_user = User(id=self._next_id, detail=data)
            self._users[new_user.id] = new_user
            self._next_id += 1
            return new_user
        elif isinstance(data, User):
            self._users[data.id] = data
            return data
        else:
            raise TypeError("Unsupported type for put method")

class UserService:
    def __init__(self, user_repository: UserRepositoryProtocol):
        self._user_repository = user_repository

    def create_user(self, detail: UserDetail) -> User:
        return self._user_repository.put(detail)

    def update_user(self, user_id: int, detail: UserDetail) -> Optional[User]:
        user = self._user_repository.fetch(user_id)
        if not user:
            return None

        user.detail.name = detail.name
        user.detail.email = detail.email
        return self._user_repository.put(user)

# --- DI Container ---

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    user_repository = providers.Singleton(
        UserRepositoryOnMemory
    )

    user_service = providers.Factory(
        UserService,
        user_repository=user_repository,
    )

# --- Application Execution ---

def main():
    container = Container()
    user_service = container.user_service()

    created_user = user_service.create_user(UserDetail(name="田中太郎", email="[email protected]"))
    print(f"作成されたユーザー: {created_user}")

    updated_user = user_service.update_user(
        created_user.id, UserDetail(name="田中次郎", email="[email protected]")
    )
    print(f"更新されたユーザー: {updated_user}")

if __name__ == "__main__":
    main()

# --- Tests ---

class TestUserService:
    def test_create_user(self):
        mock_repository = Mock(spec=UserRepositoryProtocol)
        detail = UserDetail(name="新規ユーザー", email="[email protected]")
        saved_user = User(id=1, detail=detail)
        mock_repository.put.return_value = saved_user

        user_service = UserService(mock_repository)
        result = user_service.create_user(detail)

        assert result == saved_user
        mock_repository.put.assert_called_once_with(detail)

    def test_update_user_success(self):
        mock_repository = Mock(spec=UserRepositoryProtocol)
        detail = UserDetail(name="新ユーザー", email="[email protected]")
        existing_user = User(id=1, detail=UserDetail(name="旧ユーザー", email="[email protected]"))
        mock_repository.fetch.return_value = existing_user

        updated_user = User(id=1, detail=detail)
        mock_repository.put.return_value = updated_user

        user_service = UserService(mock_repository)
        result = user_service.update_user(1, detail)

        assert result == updated_user
        mock_repository.fetch.assert_called_once_with(1)
        mock_repository.put.assert_called_once_with(updated_user)

    def test_update_user_not_found(self):
        mock_repository = Mock(spec=UserRepositoryProtocol)
        mock_repository.fetch.return_value = None

        user_service = UserService(mock_repository)
        result = user_service.update_user(99, UserDetail(name="誰か", email="[email protected]"))

        assert result is None
        mock_repository.fetch.assert_called_once_with(99)
        mock_repository.put.assert_not_called()

main2.py

from typing import Protocol, Optional, Union
from dataclasses import dataclass
from dependency_injector import containers, providers
import pytest
from unittest.mock import Mock

# --- Data Models ---

@dataclass
class UserDetail:
    name: str
    email: str

@dataclass
class User:
    id: int
    detail: UserDetail

# --- Protocols (Interfaces) ---

class UserRepositoryProtocol(Protocol):
    def fetch(self, user_id: int) -> Optional[User]:
        ...

    def put(self, data: Union[User, UserDetail]) -> User:
        ...

# --- Concrete Implementations ---

class UserRepositoryOnMemory:
    def __init__(self):
        self._users: dict[int, User] = {}
        self._next_id = 1

    def fetch(self, user_id: int) -> Optional[User]:
        return self._users.get(user_id)

    def put(self, data: Union[User, UserDetail]) -> User:
        if isinstance(data, UserDetail):
            new_user = User(id=self._next_id, detail=data)
            self._users[new_user.id] = new_user
            self._next_id += 1
            return new_user
        elif isinstance(data, User):
            self._users[data.id] = data
            return data
        else:
            raise TypeError("Unsupported type for put method")

class UserService:
    def __init__(self, user_repository: UserRepositoryProtocol):
        self._user_repository = user_repository

    def create_user(self, detail: UserDetail) -> User:
        return self._user_repository.put(detail)

    def update_user(self, user_id: int, detail: UserDetail) -> Optional[User]:
        user = self._user_repository.fetch(user_id)
        if not user:
            return None

        user.detail.name = detail.name
        user.detail.email = detail.email
        return self._user_repository.put(user)

# --- DI Container ---

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    user_repository = providers.Singleton(
        UserRepositoryOnMemory
    )

    user_service = providers.Factory(
        UserService,
        user_repository=user_repository,
    )

# --- Application Execution ---

def main():
    container = Container()
    user_service = container.user_service()

    created_user = user_service.create_user(UserDetail(name="田中太郎", email="[email protected]"))
    print(f"作成されたユーザー: {created_user}")

    updated_user = user_service.update_user(
        created_user.id, UserDetail(name="田中次郎", email="[email protected]")
    )
    print(f"更新されたユーザー: {updated_user}")

if __name__ == "__main__":
    main()

# --- Tests ---

class UserRepositoryFixture:
    def __init__(self):
        self._predefined_users: dict[int, User] = {}
        self._next_id = 1
        self._fetch_calls: list[int] = []
        self._put_calls: list[Union[User, UserDetail]] = []

    def set_user(self, user: User):
        """テスト用にユーザーを事前に設定"""
        self._predefined_users[user.id] = user

    def fetch(self, user_id: int) -> Optional[User]:
        self._fetch_calls.append(user_id)
        return self._predefined_users.get(user_id)

    def put(self, data: Union[User, UserDetail]) -> User:
        self._put_calls.append(data)
        if isinstance(data, UserDetail):
            new_user = User(id=self._next_id, detail=data)
            self._predefined_users[new_user.id] = new_user
            self._next_id += 1
            return new_user
        elif isinstance(data, User):
            self._predefined_users[data.id] = data
            return data
        else:
            raise TypeError("Unsupported type for put method")

    def get_fetch_calls(self) -> list[int]:
        """fetchが呼ばれた引数の履歴を取得"""
        return self._fetch_calls.copy()

    def get_put_calls(self) -> list[Union[User, UserDetail]]:
        """putが呼ばれた引数の履歴を取得"""
        return self._put_calls.copy()

class TestUserServiceWithFixture:
    def test_create_user(self):
        test_repository = UserRepositoryFixture()
        detail = UserDetail(name="新規ユーザー", email="[email protected]")

        user_service = UserService(test_repository)
        result = user_service.create_user(detail)

        assert result.id == 1
        assert result.detail == detail
        assert test_repository.get_put_calls() == [detail]

    def test_update_user_success(self):
        test_repository = UserRepositoryFixture()
        detail = UserDetail(name="新ユーザー", email="[email protected]")
        existing_user = User(id=1, detail=UserDetail(name="旧ユーザー", email="[email protected]"))
        test_repository.set_user(existing_user)

        user_service = UserService(test_repository)
        result = user_service.update_user(1, detail)

        assert result is not None
        assert result.id == 1
        assert result.detail.name == "新ユーザー"
        assert result.detail.email == "[email protected]"
        assert test_repository.get_fetch_calls() == [1]
        assert len(test_repository.get_put_calls()) == 1

    def test_update_user_not_found(self):
        test_repository = UserRepositoryFixture()

        user_service = UserService(test_repository)
        result = user_service.update_user(99, UserDetail(name="誰か", email="[email protected]"))

        assert result is None
        assert test_repository.get_fetch_calls() == [99]
        assert test_repository.get_put_calls() == []