クリーンアーキテクチャとは何か、そしてなぜPythonにおいて重要なのか？

クリーンアーキテクチャとは、関心事項をレイヤーごとに分離し、アプリケーションをフレームワークやデータベース、UIから独立させるソフトウェア設計の哲学です。Pythonでは、ビジネスロジックを中心にコードを整理することにより、テスト可能なコードの実現、保守の容易化、フレームワークの柔軟性、およびチーム間の協業の向上が可能になります。

SOLID原則とは何か、そしてそれらがPythonにおいてどのように適用されるか。

SOLID原則は、単一責任、オープン/クローズド、リスコブ・サブスティテーション、インターフェース分離、依存関係逆転の5つの設計ガイドラインです。Pythonでは、ダックタイピング、抽象基底クラス、プロトコル、依存性注入を用いてクラス設計を行い、テストや拡張が容易な柔軟で保守性の高いコードを実現します。

Pythonでリポジトリパターンをどのように実装するか？

リポジトリパターンは、データベース操作をカプセル化したクラスを使用してデータアクセスを抽象化します。CRUDメソッドを持つ抽象基底クラスを定義し、SQL、NoSQL、APIなどの具体的なデータソース向けにリポジトリを実装し、サービス層にリポジトリを注入します。これにより、ビジネスロジックとデータストレージが分離されます。

六角形アーキテクチャと層状アーキテクチャの違いは何か？

レイヤードアーキテクチャは、プレゼンテーション、ビジネス、データの水平層にコードを整理し、依存関係は下方向に流れる。ヘキサゴナルアーキテクチャ（ポートとアダプター）は、ビジネスロジックを中央に配置し、その周りにアダプターを配置することで、すべての外部依存関係を交換可能にする。どちらもPythonで動作するが、ヘキサゴナルアーキテクチャはテスト性がより高い。

ドメインモデルではPydanticとdataclassesのどちらを使用すべきでしょうか？

基本的な検証と不変性が必要なシンプルなドメインモデルでは、Pythonのdataclassesを使用してください。高度な検証、シリアライズ、API統合、またはFastAPIでの作業が必要な場合はPydanticを選びましょう。Pydanticはランタイムでの検証に優れており、一方でdataclassesは内部のドメインオブジェクトに対して軽量で高速です。

Pythonでフレームワークを使わずに依存関係注入をどのように実装するか？

Pythonはコンストラクタベースの依存関係注入を自然にサポートしています。依存関係をコンストラクタのパラメータとして渡し、抽象化にはプロトコルやABC（抽象基底クラス）を使用し、必要に応じて単純なファクトリーやサービスロケータを活用してください。複雑なアプリケーションでは、dependency-injectorやpunqなどの軽量なライブラリを使用し、重いフレームワークを避けることを検討してください。

クリーンアーキテクチャ向けのPythonデザインパターン

SOLID設計パターンを使って、保守性の高いPythonアプリケーションを構築しましょう。

Clean Architectureは、関心の分離と依存関係の管理を強調することで、開発者がスケーラブルで保守可能なアプリケーションを構築する方法を革命的に変えてきました。

Pythonでは、これらの原則が言語の動的性質と組み合わさり、ビジネス要件に応じて進化しながらも技術的負債にならない柔軟でテスト可能なシステムを構築できます。

活気のあるテクノロジー会議ホール

PythonにおけるClean Architectureの理解

Clean Architectureは、Robert C. Martin（Uncle Bob）によって導入され、ソフトウェアを同心円状のレイヤーに組織化し、依存関係がコアなビジネスロジックに向かって内側を指すようにします。このアーキテクチャパターンは、アプリケーションの重要なビジネスルールがフレームワーク、データベース、外部サービスに依存しないことを保証します。

核心的な哲学

基本的な原則は単純ですが強力です：ビジネスロジックはインフラストラクチャに依存してはいけません。あなたのドメインエンティティ、ユースケース、ビジネスルールは、PostgreSQLを使用しているかMongoDBを使用しているか、FastAPIを使用しているかFlaskを使用しているか、AWSを使用しているかAzureを使用しているかに関係なく動作する必要があります。

Pythonでは、この哲学が言語の「ダックタイピング」およびプロトコル指向プログラミングと完璧に一致しており、静的に型付けされた言語で必要な儀礼を伴わずにクリーンな分離を実現できます。

Clean Architectureの4つのレイヤー

エンティティレイヤー（ドメイン）：企業全体にわたるビジネスルールを持つ純粋なビジネスオブジェクト。これらは外部依存がないPOJO（Plain Old Python Objects）です。

ユースケースレイヤー（アプリケーション）：エンティティと外部サービスの間でデータの流れを調整するアプリケーション固有のビジネスルール。

インターフェースアダプタレイヤー：ユースケースとエンティティにとって最も都合の良い形式と、外部機関が要求する形式の間でデータを変換します。

フレームワーク＆ドライバレイヤー：データベース、ウェブフレームワーク、外部APIなどのすべての外部詳細。

PythonにおけるSOLID原則

SOLID原則はクリーンアーキテクチャの基礎です。それぞれの原則がPythonでどのように現れるかを見てみましょう。Pythonにおけるデザインパターンの包括的な概要については、Pythonデザインパターンガイドを参照してください。

単一責任の原則（SRP）

各クラスは変更の理由が1つだけである必要があります：

# バッド：複数の責任
class UserManager:
    def create_user(self, user_data):
        # ユーザー作成
        pass
    
    def send_welcome_email(self, user):
        # メール送信
        pass
    
    def log_creation(self, user):
        # ファイルへのログ記録
        pass

# グッド：責任の分離
class UserService:
    def __init__(self, repository, email_service, logger):
        self.repository = repository
        self.email_service = email_service
        self.logger = logger
    
    def create_user(self, user_data):
        user = User(**user_data)
        self.repository.save(user)
        self.email_service.send_welcome(user)
        self.logger.info(f"User created: {user.id}")
        return user

開放/閉鎖の原則（OCP）

ソフトウェアエンティティは拡張に対して開かれていて、変更に対して閉じている必要があります：

from abc import ABC, abstractmethod
from typing import Protocol

# プロトコルを使用（Python 3.8+）
class PaymentProcessor(Protocol):
    def process_payment(self, amount: float) -> bool:
        ...

class CreditCardProcessor:
    def process_payment(self, amount: float) -> bool:
        # クレジットカードロジック
        return True

class PayPalProcessor:
    def process_payment(self, amount: float) -> bool:
        # PayPalロジック
        return True

# 既存のコードを変更せずに簡単に拡張可能
class CryptoProcessor:
    def process_payment(self, amount: float) -> bool:
        # クリプト通貨ロジック
        return True

リスコフの置換の原則（LSP）

オブジェクトはサブタイプで置き換えられてもプログラムが壊れない必要があります：

from abc import ABC, abstractmethod

class DataStore(ABC):
    @abstractmethod
    def save(self, key: str, value: str) -> None:
        pass
    
    @abstractmethod
    def get(self, key: str) -> str:
        pass

class PostgreSQLStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # PostgreSQL実装
        pass
    
    def get(self, key: str) -> str:
        # PostgreSQL実装
        return ""

class RedisStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Redis実装
        pass
    
    def get(self, key: str) -> str:
        # Redis実装
        return ""

# どちらも相互に使用可能
def process_data(store: DataStore, key: str, value: str):
    store.save(key, value)
    return store.get(key)

インターフェース分離の原則（ISP）

クライアントは使用していないインターフェースに依存してはいけません：

# バッド：太いインターフェース
class Worker(ABC):
    @abstractmethod
    def work(self): pass
    
    @abstractmethod
    def eat(self): pass
    
    @abstractmethod
    def sleep(self): pass

# グッド：インターフェースの分離
class Workable(Protocol):
    def work(self) -> None: ...

class Eatable(Protocol):
    def eat(self) -> None: ...

class Human:
    def work(self) -> None:
        print("Working")
    
    def eat(self) -> None:
        print("Eating")

class Robot:
    def work(self) -> None:
        print("Working")
    # eatメソッドは不要

依存性逆転の原則（DIP）

高レベルモジュールは低レベルモジュールに依存してはいけません。どちらも抽象化に依存する必要があります：

from typing import Protocol

# 抽象化
class EmailSender(Protocol):
    def send(self, to: str, subject: str, body: str) -> None:
        ...

# 低レベルモジュール
class SMTPEmailSender:
    def send(self, to: str, subject: str, body: str) -> None:
        # SMTP実装
        pass

# 高レベルモジュールは抽象化に依存
class UserRegistrationService:
    def __init__(self, email_sender: EmailSender):
        self.email_sender = email_sender
    
    def register(self, email: str, name: str):
        # 登録ロジック
        self.email_sender.send(
            to=email,
            subject="Welcome!",
            body=f"Hello {name}"
        )

レポジトリパターン：データアクセスの抽象化

レポジトリパターンは、ドメインオブジェクトにアクセスするためのコレクションのようなインターフェースを提供し、データストレージの詳細を隠蔽します。

基本的なレポジトリ実装

from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4

@dataclass
class User:
    id: UUID
    email: str
    name: str
    is_active: bool = True

class UserRepository(ABC):
    @abstractmethod
    def save(self, user: User) -> User:
        pass
    
    @abstractmethod
    def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass
    
    @abstractmethod
    def get_by_email(self, email: str) -> Optional[User]:
        pass
    
    @abstractmethod
    def list_all(self) -> List[User]:
        pass
    
    @abstractmethod
    def delete(self, user_id: UUID) -> bool:
        pass

SQLAlchemy実装

from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

Base = declarative_base()

class UserModel(Base):
    __tablename__ = 'users'
    
    id = Column(PGUUID(as_uuid=True), primary_key=True)
    email = Column(String, unique=True, nullable=False)
    name = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)

class SQLAlchemyUserRepository(UserRepository):
    def __init__(self, session: Session):
        self.session = session
    
    def save(self, user: User) -> User:
        user_model = UserModel(
            id=user.id,
            email=user.email,
            name=user.name,
            is_active=user.is_active
        )
        self.session.add(user_model)
        self.session.commit()
        return user
    
    def get_by_id(self, user_id: UUID) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).first()
        
        if not user_model:
            return None
        
        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )
    
    def get_by_email(self, email: str) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.email == email
        ).first()
        
        if not user_model:
            return None
        
        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )
    
    def list_all(self) -> List[User]:
        users = self.session.query(UserModel).all()
        return [
            User(
                id=u.id,
                email=u.email,
                name=u.name,
                is_active=u.is_active
            )
            for u in users
        ]
    
    def delete(self, user_id: UUID) -> bool:
        result = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).delete()
        self.session.commit()
        return result > 0

テスト用のインメモリレポジトリ

class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self.users: dict[UUID, User] = {}
    
    def save(self, user: User) -> User:
        self.users[user.id] = user
        return user
    
    def get_by_id(self, user_id: UUID) -> Optional[User]:
        return self.users.get(user_id)
    
    def get_by_email(self, email: str) -> Optional[User]:
        for user in self.users.values():
            if user.email == email:
                return user
        return None
    
    def list_all(self) -> List[User]:
        return list(self.users.values())
    
    def delete(self, user_id: UUID) -> bool:
        if user_id in self.users:
            del self.users[user_id]
            return True
        return False

サービスレイヤー：ビジネスロジックのオーケストレーション

サービスレイヤーは、リポジトリ、外部サービス、ドメインロジックの間の流れをオーケストレーションするユースケースを実装します。

from typing import Optional
from uuid import uuid4

class UserAlreadyExistsError(Exception):
    pass

class UserNotFoundError(Exception):
    pass

class UserService:
    def __init__(
        self,
        user_repository: UserRepository,
        email_service: EmailSender,
        event_publisher: 'EventPublisher'
    ):
        self.user_repository = user_repository
        self.email_service = email_service
        self.event_publisher = event_publisher
    
    def register_user(self, email: str, name: str) -> User:
        # ユーザーが存在するか確認
        existing_user = self.user_repository.get_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(f"メール {email} に該当するユーザーは既に存在します")
        
        # 新しいユーザーを作成
        user = User(
            id=uuid4(),
            email=email,
            name=name,
            is_active=True
        )
        
        # レポジトリに保存
        user = self.user_repository.save(user)
        
        # ウェルカムメールを送信
        self.email_service.send(
            to=user.email,
            subject="Welcome!",
            body=f"こんにちは {user.name}、私たちのプラットフォームへようこそ！"
        )
        
        # イベントを発行
        self.event_publisher.publish('user.registered', {
            'user_id': str(user.id),
            'email': user.email
        })
        
        return user
    
    def deactivate_user(self, user_id: UUID) -> User:
        user = self.user_repository.get_by_id(user_id)
        if not user:
            raise UserNotFoundError(f"ユーザー {user_id} が見つかりません")
        
        user.is_active = False
        user = self.user_repository.save(user)
        
        self.event_publisher.publish('user.deactivated', {
            'user_id': str(user.id)
        })
        
        return user

Pythonにおける依存性注入

Pythonの動的性質により、重いフレームワークを必要とせずに依存性注入が簡単に実行できます。

コンストラクタ注入

class OrderService:
    def __init__(
        self,
        order_repository: 'OrderRepository',
        payment_processor: PaymentProcessor,
        notification_service: 'NotificationService'
    ):
        self.order_repository = order_repository
        self.payment_processor = payment_processor
        self.notification_service = notification_service
    
    def place_order(self, order_data: dict):
        # 注入された依存性を使用
        pass

シンプルな依存性コンテナ

from typing import Dict, Type, Callable, Any

class Container:
    def __init__(self):
        self._services: Dict[Type, Callable] = {}
        self._singletons: Dict[Type, Any] = {}
    
    def register(self, interface: Type, factory: Callable):
        self._services[interface] = factory
    
    def register_singleton(self, interface: Type, instance: Any):
        self._singletons[interface] = instance
    
    def resolve(self, interface: Type):
        if interface in self._singletons:
            return self._singletons[interface]
        
        factory = self._services.get(interface)
        if factory:
            return factory(self)
        
        raise ValueError(f"{interface} に対する登録が見つかりません")

# 使用例
def create_container() -> Container:
    container = Container()
    
    # サービスを登録
    container.register_singleton(
        Session,
        sessionmaker(bind=create_engine('postgresql://...'))()
    )
    
    container.register(
        UserRepository,
        lambda c: SQLAlchemyUserRepository(c.resolve(Session))
    )
    
    container.register(
        EmailSender,
        lambda c: SMTPEmailSender()
    )
    
    container.register(
        UserService,
        lambda c: UserService(
            c.resolve(UserRepository),
            c.resolve(EmailSender),
            c.resolve(EventPublisher)
        )
    )
    
    return container

ハーケンアーキテクチャ（ポートとアダプタ）

ハーケンアーキテクチャは、外部通信を処理するアダプタを介してビジネスロジックを中央に配置します。

ポート（インターフェース）の定義

# 入力ポート（プライマリ）
class CreateUserUseCase(Protocol):
    def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
        ...

# 出力ポート（セカンダリ）
class UserPersistencePort(Protocol):
    def save(self, user: User) -> User:
        ...
    
    def find_by_email(self, email: str) -> Optional[User]:
        ...

アダプタの実装

from pydantic import BaseModel, EmailStr

# 入力アダプタ（REST API）
from fastapi import FastAPI, Depends, HTTPException

class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str

class CreateUserResponse(BaseModel):
    id: str
    email: str
    name: str

app = FastAPI()

@app.post("/users", response_model=CreateUserResponse)
def create_user(
    request: CreateUserRequest,
    user_service: UserService = Depends(get_user_service)
):
    try:
        user = user_service.register_user(
            email=request.email,
            name=request.name
        )
        return CreateUserResponse(
            id=str(user.id),
            email=user.email,
            name=user.name
        )
    except UserAlreadyExistsError as e:
        raise HTTPException(status_code=400, detail=str(e))

# 出力アダプタ（データベース）
# SQLAlchemyUserRepositoryとして既に実装済み

ドメイン駆動設計パターン

バリューオブジェクト

属性によって定義される不変オブジェクト：

from dataclasses import dataclass
from typing import Pattern
import re

@dataclass(frozen=True)
class Email:
    value: str
    
    EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
    
    def __post_init__(self):
        if not self.EMAIL_PATTERN.match(self.value):
            raise ValueError(f"無効なメール: {self.value}")
    
    def __str__(self):
        return self.value

@dataclass(frozen=True)
class Money:
    amount: float
    currency: str
    
    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("金額は負数であってはいけません")
        if self.currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError(f"サポートされていない通貨: {self.currency}")
    
    def add(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise ValueError("異なる通貨を加算することはできません")
        return Money(self.amount + other.amount, self.currency)

アグリゲート

ドメインオブジェクトのクラスタで、単一の単位として扱われます：

from dataclasses import dataclass, field
from typing import List
from datetime import datetime

@dataclass
class OrderItem:
    product_id: UUID
    quantity: int
    price: Money
    
    def total(self) -> Money:
        return Money(
            self.price.amount * self.quantity,
            self.price.currency
        )

@dataclass
class Order:
    id: UUID
    customer_id: UUID
    items: List[OrderItem] = field(default_factory=list)
    status: str = "pending"
    created_at: datetime = field(default_factory=datetime.now)
    
    def add_item(self, product_id: UUID, quantity: int, price: Money):
        item = OrderItem(product_id, quantity, price)
        self.items.append(item)
    
    def remove_item(self, product_id: UUID):
        self.items = [
            item for item in self.items 
            if item.product_id != product_id
        ]
    
    def total(self) -> Money:
        if not self.items:
            return Money(0, "USD")
        
        return sum(
            (item.total() for item in self.items),
            Money(0, self.items[0].price.currency)
        )
    
    def confirm(self):
        if not self.items:
            raise ValueError("空の注文は確認できません")
        if self.status != "pending":
            raise ValueError("注文はすでに処理されています")
        self.status = "confirmed"

ドメインイベント

ドメインイベントは、コンポーネント間の疎結合を可能にし、イベント駆動型アーキテクチャをサポートします。本格的なイベント駆動型システムを構築するには、AWS Kinesisなどのサービスを使用してイベントストリーミングを実装することを検討してください。AWS Kinesisを使用したイベント駆動型マイクロサービスの構築を参照してください。

from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable

@dataclass
class DomainEvent:
    occurred_at: datetime = field(default_factory=datetime.now)

@dataclass
class OrderConfirmed(DomainEvent):
    order_id: UUID
    customer_id: UUID
    total: Money

class EventPublisher:
    def __init__(self):
        self._handlers: Dict[Type, List[Callable]] = {}
    
    def subscribe(self, event_type: Type, handler: Callable):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)
    
    def publish(self, event: DomainEvent):
        event_type = type(event)
        handlers = self._handlers.get(event_type, [])
        for handler in handlers:
            handler(event)

モダンな Python の特徴とクリーンアーキテクチャ

Python のモダンな特徴は、クリーンアーキテクチャを実装する際により洗練され、型安全な実装が可能になります。Python の構文や特徴に関するクイックリファレンスが必要な場合は、Python Cheatsheet をご確認ください。

型ヒントとプロトコル

from typing import Protocol, runtime_checkable

@runtime_checkable
class Serializable(Protocol):
    def to_dict(self) -> dict:
        ...
    
    @classmethod
    def from_dict(cls, data: dict) -> 'Serializable':
        ...

def serialize(obj: Serializable) -> dict:
    return obj.to_dict()

Pydantic を用いた検証

from pydantic import BaseModel, Field, validator
from typing import Optional

class CreateUserDTO(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2, max_length=100)
    age: Optional[int] = Field(None, ge=0, le=150)
    
    @validator('name')
    def name_must_not_contain_numbers(cls, v):
        if any(char.isdigit() for char in v):
            raise ValueError('Name cannot contain numbers')
        return v
    
    class Config:
        frozen = True  # 不変性を設定

I/O 操作用の Async/Await

Python の async/await 構文は、クリーンアーキテクチャにおける I/O バウンド操作において特に強力で、データベースや外部サービスとの非ブロッキングなインタラクションを可能にします。Python アプリケーションをサーバーレスプラットフォームにデプロイする際には、パフォーマンス特性を理解することが重要です。最適化に関する洞察については、AWS lambda performance: JavaScript vs Python vs Golang をご参照ください。

from typing import List
import asyncio

class AsyncUserRepository(ABC):
    @abstractmethod
    async def save(self, user: User) -> User:
        pass
    
    @abstractmethod
    async def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass

class AsyncUserService:
    def __init__(self, repository: AsyncUserRepository):
        self.repository = repository
    
    async def register_user(self, email: str, name: str) -> User:
        user = User(id=uuid4(), email=email, name=name)
        return await self.repository.save(user)
    
    async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
        tasks = [self.repository.get_by_id(uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
        return [u for u in results if u is not None]

プロジェクト構造のベストプラクティス

クリーンアーキテクチャを維持するためには、適切なプロジェクト構成が不可欠です。プロジェクト構造を設定する前に、依存関係の分離のために Python の仮想環境を使用することを確認してください。venv Cheatsheet は仮想環境の管理に関するすべての知識を網羅しています。現代的な Python プロジェクトでは、uv - New Python Package, Project, and Environment Manager の使用を検討してください。これはパッケージ管理とプロジェクトのセットアップをより高速に行うことができます。

my_application/
├── domain/                 # 企業のビジネスルール
│   ├── __init__.py
│   ├── entities/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── order.py
│   ├── value_objects/
│   │   ├── __init__.py
│   │   ├── email.py
│   │   └── money.py
│   ├── events/
│   │   ├── __init__.py
│   │   └── user_events.py
│   └── exceptions.py
│
├── application/            # アプリケーションのビジネスルール
│   ├── __init__.py
│   ├── use_cases/
│   │   ├── __init__.py
│   │   ├── create_user.py
│   │   └── place_order.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── ports/
│       ├── __init__.py
│       ├── repositories.py
│       └── external_services.py
│
├── infrastructure/         # 外部インターフェース
│   ├── __init__.py
│   ├── persistence/
│   │   ├── __init__.py
│   │   ├── sqlalchemy/
│   │   │   ├── models.py
│   │   │   └── repositories.py
│   │   └── mongodb/
│   │       └── repositories.py
│   ├── messaging/
│   │   ├── __init__.py
│   │   └── rabbitmq_publisher.py
│   ├── external_services/
│   │   ├── __init__.py
│   │   └── email_service.py
│   └── config.py
│
├── presentation/           # UI/API レイヤー
│   ├── __init__.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── dependencies.py
│   │   ├── routes/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   └── orders.py
│   │   └── schemas/
│   │       ├── __init__.py
│   │       └── user_schemas.py
│   └── cli/
│       └── commands.py
│
├── tests/
│   ├── unit/
│   ├── integration/
│   └── e2e/
│
├── main.py                 # アプリケーションのエントリーポイント
├── container.py            # 依存性注入の設定
├── pyproject.toml
└── README.md

クリーンアーキテクチャのテスト

ドメインロジックのユニットテスト

import pytest
from uuid import uuid4

def test_user_creation():
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )
    assert user.email == "test@example.com"
    assert user.is_active is True

def test_order_total_calculation():
    order = Order(id=uuid4(), customer_id=uuid4())
    order.add_item(
        uuid4(),
        quantity=2,
        price=Money(10.0, "USD")
    )
    order.add_item(
        uuid4(),
        quantity=1,
        price=Money(5.0, "USD")
    )
    assert order.total().amount == 25.0

レポジトリとの統合テスト

@pytest.fixture
def in_memory_repository():
    return InMemoryUserRepository()

def test_user_repository_save_and_retrieve(in_memory_repository):
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )
    
    saved_user = in_memory_repository.save(user)
    retrieved_user = in_memory_repository.get_by_id(user.id)
    
    assert retrieved_user is not None
    assert retrieved_user.email == user.email

サービス層のテスト

from unittest.mock import Mock

def test_user_registration():
    # Arrange
    mock_repository = Mock(spec=UserRepository)
    mock_repository.get_by_email.return_value = None
    mock_repository.save.return_value = User(
        id=uuid4(),
        email="test@example.com",
        name="Test"
    )
    
    mock_email = Mock(spec=EmailSender)
    mock_events = Mock(spec=EventPublisher)
    
    service = UserService(mock_repository, mock_email, mock_events)
    
    # Act
    user = service.register_user("test@example.com", "Test")
    
    # Assert
    assert user.email == "test@example.com"
    mock_repository.save.assert_called_once()
    mock_email.send.assert_called_once()
    mock_events.publish.assert_called_once()

一般的な落とし穴と回避方法

過剰設計

シンプルな CRUD アプリケーションのためにクリーンアーキテクチャを実装しないでください。シンプルに始め、複雑さが増すにつれてリファクタリングしてください。

漏洩した抽象化

ドメインエンティティがデータベースの注釈やフレームワーク固有のコードを含まないことを確認してください：

# Bad
from sqlalchemy import Column

@dataclass
class User:
    id: Column(Integer, primary_key=True)  # フレームワークがドメインに漏れている

# Good
@dataclass
class User:
    id: UUID  # 純粋なドメインオブジェクト

循環依存

層間の循環依存を解消するために、依存性注入とインターフェースを使用してください。

コンテキストの無視

クリーンアーキテクチャは万能ではありません。プロジェクトの規模とチームの専門性に応じて、層の厳密性を調整してください。