Python パターン

このスキルは、一般的な設計原則を Python 固有のイディオムで拡張した包括的な Python パターンを提供します。

Protocol (ダック タイピング)

構造的部分型付け (型ヒント付きダックタイピング) には Protocol を使用します:

from typing import Protocol

class Repository(Protocol):
    def find_by_id(self, id: str) -> dict | None: ...
    def save(self, entity: dict) -> dict: ...

# これらのメソッドを持つすべてのクラスがプロトコルを満たします
class UserRepository:
    def find_by_id(self, id: str) -> dict | None:
        # 実装
        pass

    def save(self, entity: dict) -> dict:
        # 実装
        pass

def process_entity(repo: Repository, id: str) -> None:
    entity = repo.find_by_id(id)
    # ... 処理

メリット:

• 継承なしの型安全性
• 柔軟で疎結合なコード
• テストとモックが容易

データクラスを DTO として使用

データ転送オブジェクトと値オブジェクトには dataclass を使用します:

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class CreateUserRequest:
    name: str
    email: str
    age: Optional[int] = None
    tags: list[str] = field(default_factory=list)

@dataclass(frozen=True)
class User:
    """イミュータブルなユーザーエンティティ"""
    id: str
    name: str
    email: str

機能:

• 自動生成される __init__, __repr__, __eq__
• イミュータビリティのための frozen=True
• 複雑なデフォルト値のための field()
• 検証のための型ヒント

コンテキスト マネージャー

リソース管理には context manager (with ステートメント) を使用します:

from contextlib import contextmanager
from typing import Generator

@contextmanager
def database_transaction(db) -> Generator[None, None, None]:
    """データベース トランザクション用のコンテキスト マネージャー"""
    try:
        yield
        db.commit()
    except Exception:
        db.rollback()
        raise

# 使用例
with database_transaction(db):
    db.execute("INSERT INTO users ...")

クラスベースのコンテキスト マネージャー:

class FileProcessor:
    def __init__(self, filename: str):
        self.filename = filename
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, 'r')
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()
        return False  # 例外を抑制しない

ジェネレータ

遅延評価とメモリ効率的なイテレーションには generator を使用します:

def read_large_file(filename: str):
    """大きなファイルを 1 行ずつ読むジェネレータ"""
    with open(filename, 'r') as f:
        for line in f:
            yield line.strip()

# メモリ効率的な処理
for line in read_large_file('huge.txt'):
    process(line)

ジェネレータ式:

# リスト内包表記の代わりに
squares = (x**2 for x in range(1000000))  # 遅延評価

# パイプラインパターン
numbers = (x for x in range(100))
evens = (x for x in numbers if x % 2 == 0)
squares = (x**2 for x in evens)

デコレータ

関数デコレータ

from functools import wraps
import time

def timing(func):
    """実行時間を計測するデコレータ"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} took {end - start:.2f}s")
        return result
    return wrapper

@timing
def slow_function():
    time.sleep(1)

クラス デコレータ

def singleton(cls):
    """クラスをシングルトンにするデコレータ"""
    instances = {}

    @wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return get_instance

@singleton
class Config:
    pass

Async/Await

Async 関数

import asyncio
from typing import List

async def fetch_user(user_id: str) -> dict:
    """I/O バウンド操作用の async 関数"""
    await asyncio.sleep(0.1)  # ネットワーク呼び出しをシミュレート
    return {"id": user_id, "name": "Alice"}

async def fetch_all_users(user_ids: List[str]) -> List[dict]:
    """asyncio.gather による並行実行"""
    tasks = [fetch_user(uid) for uid in user_ids]
    return await asyncio.gather(*tasks)

# async コードを実行
asyncio.run(fetch_all_users(["1", "2", "3"]))

Async コンテキスト マネージャー

class AsyncDatabase:
    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.disconnect()

async with AsyncDatabase() as db:
    await db.query("SELECT * FROM users")

型ヒント

高度な型ヒント

from typing import TypeVar, Generic, Callable, ParamSpec, Concatenate

T = TypeVar('T')
P = ParamSpec('P')

class Repository(Generic[T]):
    """ジェネリック repository パターン"""
    def __init__(self, entity_type: type[T]):
        self.entity_type = entity_type

    def find_by_id(self, id: str) -> T | None:
        # 実装
        pass

# 型安全なデコレータ
def log_call(func: Callable[P, T]) -> Callable[P, T]:
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        print(f"Calling {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

Union 型 (Python 3.10+)

def process(value: str | int | None) -> str:
    match value:
        case str():
            return value.upper()
        case int():
            return str(value)
        case None:
            return "empty"

依存性注入

コンストラクタ注入

class UserService:
    def __init__(
        self,
        repository: Repository,
        logger: Logger,
        cache: Cache | None = None
    ):
        self.repository = repository
        self.logger = logger
        self.cache = cache

    def get_user(self, user_id: str) -> User | None:
        if self.cache:
            cached = self.cache.get(user_id)
            if cached:
                return cached

        user = self.repository.find_by_id(user_id)
        if user and self.cache:
            self.cache.set(user_id, user)

        return user

パッケージ構成

プロジェクト構造

project/
├── src/
│   └── mypackage/
│       ├── __init__.py
│       ├── domain/          # ビジネスロジック
│       │   ├── __init__.py
│       │   └── models.py
│       ├── services/        # アプリケーション サービス
│       │   ├── __init__.py
│       │   └── user_service.py
│       └── infrastructure/  # 外部依存
│           ├── __init__.py
│           └── database.py
├── tests/
│   ├── unit/
│   └── integration/
├── pyproject.toml
└── README.md

モジュール エクスポート

# __init__.py
from .models import User, Product
from .services import UserService

__all__ = ['User', 'Product', 'UserService']

エラー処理

カスタム例外

class DomainError(Exception):
    """ドメイン エラーの基本例外"""
    pass

class UserNotFoundError(DomainError):
    """ユーザーが見つからないときに発生"""
    def __init__(self, user_id: str):
        self.user_id = user_id
        super().__init__(f"User {user_id} not found")

class ValidationError(DomainError):
    """検証に失敗したときに発生"""
    def __init__(self, field: str, message: str):
        self.field = field
        self.message = message
        super().__init__(f"{field}: {message}")

例外グループ (Python 3.11+)

try:
    # 複数の操作
    pass
except* ValueError as eg:
    # すべての ValueError インスタンスを処理
    for exc in eg.exceptions:
        print(f"ValueError: {exc}")
except* TypeError as eg:
    # すべての TypeError インスタンスを処理
    for exc in eg.exceptions:
        print(f"TypeError: {exc}")

プロパティ デコレータ

class User:
    def __init__(self, name: str):
        self._name = name
        self._email = None

    @property
    def name(self) -> str:
        """読み取り専用プロパティ"""
        return self._name

    @property
    def email(self) -> str | None:
        return self._email

    @email.setter
    def email(self, value: str) -> None:
        if '@' not in value:
            raise ValueError("Invalid email")
        self._email = value

関数型プログラミング

高階関数

from functools import reduce
from typing import Callable, TypeVar

T = TypeVar('T')
U = TypeVar('U')

def pipe(*functions: Callable) -> Callable:
    """関数を左から右に合成"""
    def inner(arg):
        return reduce(lambda x, f: f(x), functions, arg)
    return inner

# 使用例
process = pipe(
    str.strip,
    str.lower,
    lambda s: s.replace(' ', '_')
)
result = process("  Hello World  ")  # "hello_world"

このスキルをいつ使うか

• Python API とパッケージの設計
• async/並行システムの実装
• Python プロジェクトの構成
• Pythonic なコードの記述
• Python コードベースのリファクタリング
• 型安全な Python 開発