python-design-patterns

Python のデザインパターン

基本的な設計原則を使って、保守しやすい Python コードを書きましょう。これらのパターンを活用すると、理解、テスト、変更が容易なシステムを構築できます。

このスキルを使う場面

• 新しいコンポーネントやサービスを設計するとき
• 複雑に絡み合ったコードをリファクタリングするとき
• 抽象化を作成するかどうかを判断するとき
• 継承とコンポジションのどちらを使うか選ぶとき
• コードの複雑さと結合度を評価するとき
• モジュール型アーキテクチャを計画するとき

基本概念

1. KISS（シンプルに保つ）

機能する最もシンプルな解決策を選びましょう。複雑さは、具体的な要件によって正当化される必要があります。

2. 単一責任の原則（SRP）

各単位には、変更する理由が1つだけあるべきです。関心事を、目的に特化したコンポーネントに分離しましょう。

3. 継承よりコンポジション

クラスを拡張するのではなく、オブジェクトを組み合わせて振る舞いを構築しましょう。

4. 3回のルール

抽象化する前に、同じものが3つ現れるまで待ちましょう。早まった抽象化より、重複のほうが適切な場合もよくあります。

クイックスタート

# Simple beats clever
# Instead of a factory/registry pattern:
FORMATTERS = {"json": JsonFormatter, "csv": CsvFormatter}

def get_formatter(name: str) -> Formatter:
    return FORMATTERS[name]()

基本パターン

パターン 1：KISS — シンプルに保つ

複雑さを加える前に、よりシンプルな解決策で機能するかどうかを考えましょう。

# Over-engineered: Factory with registration
class OutputFormatterFactory:
    _formatters: dict[str, type[Formatter]] = {}

    @classmethod
    def register(cls, name: str):
        def decorator(formatter_cls):
            cls._formatters[name] = formatter_cls
            return formatter_cls
        return decorator

    @classmethod
    def create(cls, name: str) -> Formatter:
        return cls._formatters[name]()

@OutputFormatterFactory.register("json")
class JsonFormatter(Formatter):
    ...

# Simple: Just use a dictionary
FORMATTERS = {
    "json": JsonFormatter,
    "csv": CsvFormatter,
    "xml": XmlFormatter,
}

def get_formatter(name: str) -> Formatter:
    """Get formatter by name."""
    if name not in FORMATTERS:
        raise ValueError(f"Unknown format: {name}")
    return FORMATTERS[name]()

ここでは、ファクトリーパターンを使っても価値は増えず、コードが増えるだけです。パターンは実際の問題を解決するときに使いましょう。

パターン 2：単一責任の原則

各クラスや関数には、変更する理由が1つだけあるべきです。

# BAD: Handler does everything
class UserHandler:
    async def create_user(self, request: Request) -> Response:
        # HTTP parsing
        data = await request.json()

        # Validation
        if not data.get("email"):
            return Response({"error": "email required"}, status=400)

        # Database access
        user = await db.execute(
            "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *",
            data["email"], data["name"]
        )

        # Response formatting
        return Response({"id": user.id, "email": user.email}, status=201)

# GOOD: Separated concerns
class UserService:
    """Business logic only."""

    def __init__(self, repo: UserRepository) -> None:
        self._repo = repo

    async def create_user(self, data: CreateUserInput) -> User:
        # Only business rules here
        user = User(email=data.email, name=data.name)
        return await self._repo.save(user)

class UserHandler:
    """HTTP concerns only."""

    def __init__(self, service: UserService) -> None:
        self._service = service

    async def create_user(self, request: Request) -> Response:
        data = CreateUserInput(**(await request.json()))
        user = await self._service.create_user(data)
        return Response(user.to_dict(), status=201)

これで、HTTP の変更がビジネスロジックに影響せず、その逆も同様です。

パターン 3：関心事の分離

役割が明確な独立したレイヤーにコードを整理しましょう。

┌─────────────────────────────────────────────────────┐
│  API Layer (handlers)                                │
│  - Parse requests                                    │
│  - Call services                                     │
│  - Format responses                                  │
└─────────────────────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────┐
│  Service Layer (business logic)                      │
│  - Domain rules and validation                       │
│  - Orchestrate operations                            │
│  - Pure functions where possible                     │
└─────────────────────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────┐
│  Repository Layer (data access)                      │
│  - SQL queries                                       │
│  - External API calls                                │
│  - Cache operations                                  │
└─────────────────────────────────────────────────────┘

各レイヤーが依存するのは、その下位のレイヤーだけです。

# Repository: Data access
class UserRepository:
    async def get_by_id(self, user_id: str) -> User | None:
        row = await self._db.fetchrow(
            "SELECT * FROM users WHERE id = $1", user_id
        )
        return User(**row) if row else None

# Service: Business logic
class UserService:
    def __init__(self, repo: UserRepository) -> None:
        self._repo = repo

    async def get_user(self, user_id: str) -> User:
        user = await self._repo.get_by_id(user_id)
        if user is None:
            raise UserNotFoundError(user_id)
        return user

# Handler: HTTP concerns
@app.get("/users/{user_id}")
async def get_user(user_id: str) -> UserResponse:
    user = await user_service.get_user(user_id)
    return UserResponse.from_user(user)

パターン 4：継承よりコンポジション

継承するのではなく、オブジェクトを組み合わせて振る舞いを構築しましょう。

# Inheritance: Rigid and hard to test
class EmailNotificationService(NotificationService):
    def __init__(self):
        super().__init__()
        self._smtp = SmtpClient()  # Hard to mock

    def notify(self, user: User, message: str) -> None:
        self._smtp.send(user.email, message)

# Composition: Flexible and testable
class NotificationService:
    """Send notifications via multiple channels."""

    def __init__(
        self,
        email_sender: EmailSender,
        sms_sender: SmsSender | None = None,
        push_sender: PushSender | None = None,
    ) -> None:
        self._email = email_sender
        self._sms = sms_sender
        self._push = push_sender

    async def notify(
        self,
        user: User,
        message: str,
        channels: set[str] | None = None,
    ) -> None:
        channels = channels or {"email"}

        if "email" in channels:
            await self._email.send(user.email, message)

        if "sms" in channels and self._sms and user.phone:
            await self._sms.send(user.phone, message)

        if "push" in channels and self._push and user.device_token:
            await self._push.send(user.device_token, message)

# Easy to test with fakes
service = NotificationService(
    email_sender=FakeEmailSender(),
    sms_sender=FakeSmsSender(),
)

高度なパターン

パターン 5：3回のルール

抽象化する前に、同じものが3つ現れるまで待ちましょう。

# Two similar functions? Don't abstract yet
def process_orders(orders: list[Order]) -> list[Result]:
    results = []
    for order in orders:
        validated = validate_order(order)
        result = process_validated_order(validated)
        results.append(result)
    return results

def process_returns(returns: list[Return]) -> list[Result]:
    results = []
    for ret in returns:
        validated = validate_return(ret)
        result = process_validated_return(validated)
        results.append(result)
    return results

# These look similar, but wait! Are they actually the same?
# Different validation, different processing, different errors...
# Duplication is often better than the wrong abstraction

# Only after a third case, consider if there's a real pattern
# But even then, sometimes explicit is better than abstract

パターン 6：関数のサイズに関するガイドライン

関数の役割を明確に保ちましょう。次の場合は、関数を切り出します。

• 20～50行を超える場合（複雑さによって異なります）
• 異なる複数の目的を担う場合
• ロジックのネストが深い場合（3階層以上）

# Too long, multiple concerns mixed
def process_order(order: Order) -> Result:
    # 50 lines of validation...
    # 30 lines of inventory check...
    # 40 lines of payment processing...
    # 20 lines of notification...
    pass

# Better: Composed from focused functions
def process_order(order: Order) -> Result:
    """Process a customer order through the complete workflow."""
    validate_order(order)
    reserve_inventory(order)
    payment_result = charge_payment(order)
    send_confirmation(order, payment_result)
    return Result(success=True, order_id=order.id)

パターン 7：依存性注入

テストしやすくするため、コンストラクターを通じて依存関係を渡しましょう。

from typing import Protocol

class Logger(Protocol):
    def info(self, msg: str, **kwargs) -> None: ...
    def error(self, msg: str, **kwargs) -> None: ...

class Cache(Protocol):
    async def get(self, key: str) -> str | None: ...
    async def set(self, key: str, value: str, ttl: int) -> None: ...

class UserService:
    """Service with injected dependencies."""

    def __init__(
        self,
        repository: UserRepository,
        cache: Cache,
        logger: Logger,
    ) -> None:
        self._repo = repository
        self._cache = cache
        self._logger = logger

    async def get_user(self, user_id: str) -> User:
        # Check cache first
        cached = await self._cache.get(f"user:{user_id}")
        if cached:
            self._logger.info("Cache hit", user_id=user_id)
            return User.from_json(cached)

        # Fetch from database
        user = await self._repo.get_by_id(user_id)
        if user:
            await self._cache.set(f"user:{user_id}", user.to_json(), ttl=300)

        return user

# Production
service = UserService(
    repository=PostgresUserRepository(db),
    cache=RedisCache(redis),
    logger=StructlogLogger(),
)

# Testing
service = UserService(
    repository=InMemoryUserRepository(),
    cache=FakeCache(),
    logger=NullLogger(),
)

パターン 8：よくあるアンチパターンを避ける

内部型を公開しないでください。

# BAD: Leaking ORM model to API
@app.get("/users/{id}")
def get_user(id: str) -> UserModel:  # SQLAlchemy model
    return db.query(UserModel).get(id)

# GOOD: Use response schemas
@app.get("/users/{id}")
def get_user(id: str) -> UserResponse:
    user = db.query(UserModel).get(id)
    return UserResponse.from_orm(user)

I/O とビジネスロジックを混在させないでください。

# BAD: SQL embedded in business logic
def calculate_discount(user_id: str) -> float:
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
    orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id)
    # Business logic mixed with data access

# GOOD: Repository pattern
def calculate_discount(user: User, order_history: list[Order]) -> float:
    # Pure business logic, easily testable
    if len(order_history) > 10:
        return 0.15
    return 0.0

ベストプラクティスのまとめ

1. シンプルに保つ - 機能する最もシンプルな解決策を選ぶ
2. 単一責任 - 各単位には変更する理由が1つだけある
3. 関心事を分離する - 目的が明確な独立したレイヤーに分ける
4. 継承せず、コンポーズする - 柔軟性を高めるためにオブジェクトを組み合わせる
5. 3回のルール - 抽象化は急がない
6. 関数を小さく保つ - 20～50行（複雑さによって異なります）で、目的は1つにする
7. 依存関係を注入する - テストしやすいコンストラクター注入を使う
8. 抽象化する前に削除する - 不要なコードを削除してから、パターンを検討する
9. 各レイヤーをテストする - 関心事ごとに分離してテストする
10. 気の利いたコードより明示的なコード - 優雅なコードより読みやすいコードを優先する

---

注意点

• __new__ を使ったシングルトンはサブクラスでは機能しません — サブクラスごとに独自のインスタンスが作られます。サブクラスも考慮したシングルトンが必要な場合は、__init_subclass__ またはメタクラスを使いましょう。
• データクラスの frozen=True は、ネストしたミュータブルなフィールドを凍結しません — list フィールドを持つ凍結データクラスでも、そのリストに対して .append を呼び出せます。
• オブザーバーパターンで強参照を使うと、購読者がスコープ外になってもパブリッシャーが参照を保持するため、メモリーリークが発生します。 購読者には weakref.WeakSet を使いましょう。
• インスタンスメソッドに functools.lru_cache を使うと、キャッシュが self を保持するため、インスタンスが永久に保持されます — モジュールレベルの関数に適用するか、methodtools.lru_cache を使いましょう。
• 抽象基底クラス（ABC）：@abstractmethod はサブクラスの定義時ではなく、インスタンス化時にチェックされます — 抽象メソッドを実装していないサブクラスも、インスタンスを作成しようとするまでは型チェックを通過します。