python-patterns

Python開発パターン

堅牢で効率的かつ保守性の高いアプリケーションを構築するための、Pythonらしいパターンとベストプラクティス。

いつアクティブにするか

• 新しいPythonコードを書くとき
• Pythonコードをレビューするとき
• 既存のPythonコードをリファクタリングするとき
• Pythonパッケージ/モジュールを設計するとき

中核原則

1. 可読性が重要

Pythonは可読性を優先します。コードは明白で理解しやすいものであるべきです。

# Good: 明確で読みやすい
def get_active_users(users: list[User]) -> list[User]:
    """Return only active users from the provided list."""
    return [user for user in users if user.is_active]


# Bad: 巧妙だが混乱を招く
def get_active_users(u):
    return [x for x in u if x.a]

2. 明示的であることは暗黙的であることより優れている

魔法を避け、コードが何を行っているかを明確にしましょう。

# Good: 明示的な設定
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Bad: 隠れた副作用
import some_module
some_module.setup()  # これは何をするのか？

3. EAFP - 許しを請うことは許可を得ることより簡単

Pythonは条件チェックよりも例外処理を好みます。

# Good: EAFPスタイル
def get_value(dictionary: dict, key: str, default_value: Any = None) -> Any:
    try:
        return dictionary[key]
    except KeyError:
        return default_value

# Bad: LBYL（飛び込む前に見る）スタイル
def get_value(dictionary: dict, key: str, default_value: Any = None) -> Any:
    if key in dictionary:
        return dictionary[key]
    else:
        return default_value

型ヒント

基本的な型アノテーション

from typing import Optional, List, Dict, Any

def process_user(
    user_id: str,
    data: Dict[str, Any],
    active: bool = True
) -> Optional[User]:
    """Process a user and return the updated User or None."""
    if not active:
        return None
    return User(user_id, data)

最新の型ヒント (Python 3.9+)

# Python 3.9+ - 組み込み型を使用
def process_items(items: list[str]) -> dict[str, int]:
    return {item: len(item) for item in items}

# Python 3.8 以前 - typingモジュールを使用
from typing import List, Dict

def process_items(items: List[str]) -> Dict[str, int]:
    return {item: len(item) for item in items}

型エイリアスとTypeVar

from typing import TypeVar, Union

# 複合型の型エイリアス
JSON = Union[dict[str, Any], list[Any], str, int, float, bool, None]

def parse_json(data: str) -> JSON:
    return json.loads(data)

# ジェネリック型
T = TypeVar('T')

def first(items: list[T]) -> T | None:
    """Return the first item or None if list is empty."""
    return items[0] if items else None

プロトコルベースのダックタイピング

from typing import Protocol

class Renderable(Protocol):
    def render(self) -> str:
        """Render the object to a string."""

def render_all(items: list[Renderable]) -> str:
    """Render all items that implement the Renderable protocol."""
    return "\n".join(item.render() for item in items)

エラーハンドリングパターン

特定の例外処理

# Good: 特定の例外をキャッチ
def load_config(path: str) -> Config:
    try:
        with open(path) as f:
            return Config.from_json(f.read())
    except FileNotFoundError as e:
        raise ConfigError(f"Config file not found: {path}") from e
    except json.JSONDecodeError as e:
        raise ConfigError(f"Invalid JSON in config: {path}") from e

# Bad: むき出しのexcept
def load_config(path: str) -> Config:
    try:
        with open(path) as f:
            return Config.from_json(f.read())
    except:
        return None  # 無言の失敗！

例外チェーン

def process_data(data: str) -> Result:
    try:
        parsed = json.loads(data)
    except json.JSONDecodeError as e:
        # トレースバックを保持するために例外をチェーン
        raise ValueError(f"Failed to parse data: {data}") from e

カスタム例外階層

class AppError(Exception):
    """Base exception for all application errors."""
    pass

class ValidationError(AppError):
    """Raised when input validation fails."""
    pass

class NotFoundError(AppError):
    """Raised when a requested resource is not found."""
    pass

# 使用法
def get_user(user_id: str) -> User:
    user = db.find_user(user_id)
    if not user:
        raise NotFoundError(f"User not found: {user_id}")
    return user

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

リソース管理

# Good: コンテキストマネージャーを使用
def process_file(path: str) -> str:
    with open(path, 'r') as f:
        return f.read()

# Bad: 手動リソース管理
def process_file(path: str) -> str:
    f = open(path, 'r')
    try:
        return f.read()
    finally:
        f.close()

カスタムコンテキストマネージャー

from contextlib import contextmanager

@contextmanager
def timer(name: str):
    """Context manager to time a block of code."""
    start = time.perf_counter()
    yield
    elapsed = time.perf_counter() - start
    print(f"{name} took {elapsed:.4f} seconds")

# 使用法
with timer("data processing"):
    process_large_dataset()

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

class DatabaseTransaction:
    def __init__(self, connection):
        self.connection = connection

    def __enter__(self):
        self.connection.begin_transaction()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.connection.commit()
        else:
            self.connection.rollback()
        return False  # 例外を抑制しない

# 使用法
with DatabaseTransaction(conn):
    user = conn.create_user(user_data)
    conn.create_profile(user.id, profile_data)

内包表記とジェネレーター

リスト内包表記

# Good: 単純な変換にはリスト内包表記
names = [user.name for user in users if user.is_active]

# Bad: 手動ループ
names = []
for user in users:
    if user.is_active:
        names.append(user.name)

# 複雑な内包表記は展開すべき
# Bad: 複雑すぎる
result = [x * 2 for x in items if x > 0 if x % 2 == 0]

# Good: ジェネレーター関数を使用
def filter_and_transform(items: Iterable[int]) -> list[int]:
    result = []
    for x in items:
        if x > 0 and x % 2 == 0:
            result.append(x * 2)
    return result

ジェネレーター式

# Good: 遅延評価のためのジェネレーター
total = sum(x * x for x in range(1_000_000))

# Bad: 大きな中間リストを作成
total = sum([x * x for x in range(1_000_000)])

ジェネレーター関数

def read_large_file(path: str) -> Iterator[str]:
    """Read a large file line by line."""
    with open(path) as f:
        for line in f:
            yield line.strip()

# 使用法
for line in read_large_file("huge.txt"):
    process(line)

データクラスと名前付きタプル

データクラス

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class User:
    """User entity with automatic __init__, __repr__, and __eq__."""
    id: str
    name: str
    email: str
    created_at: datetime = field(default_factory=datetime.now)
    is_active: bool = True

# 使用法
user = User(
    id="123",
    name="Alice",
    email="[email protected]"
)

バリデーション付きデータクラス

@dataclass
class User:
    email: str
    age: int

    def __post_init__(self):
        # メール形式の検証
        if "@" not in self.email:
            raise ValueError(f"Invalid email: {self.email}")
        # 年齢範囲の検証
        if self.age < 0 or self.age > 150:
            raise ValueError(f"Invalid age: {self.age}")

名前付きタプル

from typing import NamedTuple

class Point(NamedTuple):
    """Immutable 2D point."""
    x: float
    y: float

    def distance(self, other: 'Point') -> float:
        return ((self.x - other.x) ** 2 + (self.y - other.y) ** 2) ** 0.5

# 使用法
p1 = Point(0, 0)
p2 = Point(3, 4)
print(p1.distance(p2))  # 5.0

デコレーター

関数デコレーター

import functools
import time

def timer(func: Callable) -> Callable:
    """Decorator to time function execution."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

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

# slow_function() prints: slow_function took 1.0012s

パラメータ化されたデコレーター

def repeat(times: int):
    """Decorator to repeat a function multiple times."""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for _ in range(times):
                results.append(func(*args, **kwargs))
            return results
        return wrapper
    return decorator

@repeat(times=3)
def greet(name: str) -> str:
    return f"Hello, {name}!"

# greet("Alice") returns ["Hello, Alice!", "Hello, Alice!", "Hello, Alice!"]

クラスベースのデコレーター

class CountCalls:
    """Decorator that counts how many times a function is called."""
    def __init__(self, func: Callable):
        functools.update_wrapper(self, func)
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"{self.func.__name__} has been called {self.count} times")
        return self.func(*args, **kwargs)

@CountCalls
def process():
    pass

# Each call to process() prints the call count

並行性パターン

I/Oバウンドタスクのスレッド化

import concurrent.futures
import threading

def fetch_url(url: str) -> str:
    """Fetch a URL (I/O-bound operation)."""
    import urllib.request
    with urllib.request.urlopen(url) as response:
        return response.read().decode()

def fetch_all_urls(urls: list[str]) -> dict[str, str]:
    """Fetch multiple URLs concurrently using threads."""
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        future_to_url = {executor.submit(fetch_url, url): url for url in urls}
        results = {}
        for future in concurrent.futures.as_completed(future_to_url):
            url = future_to_url[future]
            try:
                results[url] = future.result()
            except Exception as e:
                results[url] = f"Error: {e}"
    return results

CPUバウンドタスクのマルチプロセッシング

def process_data(data: list[int]) -> int:
    """CPU-intensive computation."""
    return sum(x ** 2 for x in data)

def process_all(datasets: list[list[int]]) -> list[int]:
    """Process multiple datasets using multiple processes."""
    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = list(executor.map(process_data, datasets))
    return results

同時I/OのためのAsync/Await

import asyncio

async def fetch_async(url: str) -> str:
    """Fetch a URL asynchronously."""
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def fetch_all(urls: list[str]) -> dict[str, str]:
    """Fetch multiple URLs concurrently."""
    tasks = [fetch_async(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return dict(zip(urls, results))

パッケージ構成

標準的なプロジェクトレイアウト

myproject/
├── src/
│   └── mypackage/
│       ├── __init__.py
│       ├── main.py
│       ├── api/
│       │   ├── __init__.py
│       │   └── routes.py
│       ├── models/
│       │   ├── __init__.py
│       │   └── user.py
│       └── utils/
│           ├── __init__.py
│           └── helpers.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_api.py
│   └── test_models.py
├── pyproject.toml
├── README.md
└── .gitignore

インポートの慣例

# Good: インポート順 - 標準ライブラリ、サードパーティ、ローカル
import os
import sys
from pathlib import Path

import requests
from fastapi import FastAPI

from mypackage.models import User
from mypackage.utils import format_name

# Good: 自動インポートソートにはisortを使用
# pip install isort

パッケージエクスポート用の__init__.py

# mypackage/__init__.py
"""mypackage - A sample Python package."""

__version__ = "1.0.0"

# パッケージレベルで主要なクラス/関数をエクスポート
from mypackage.models import User, Post
from mypackage.utils import format_name

__all__ = ["User", "Post", "format_name"]

メモリとパフォーマンス

メモリ効率のための__slots__の使用

# Bad: 通常のクラスは__dict__を使用（メモリ大）
class Point:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

# Good: __slots__でメモリ使用量を削減
class Point:
    __slots__ = ['x', 'y']

    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

大規模データのジェネレーター

# Bad: メモリ内に完全なリストを返す
def read_lines(path: str) -> list[str]:
    with open(path) as f:
        return [line.strip() for line in f]

# Good: 行を一度に一つずつ生成

def read_lines(path: str) -> Iterator[str]:
    with open(path) as f:
        for line in f:
            yield line.strip()

ループ内での文字列連結を避ける

# Bad: 文字列の不変性によりO(n²)
result = ""
for item in items:
    result += str(item)

# Good: joinを使用してO(n)
result = "".join(str(item) for item in items)

# Good: 構築にStringIOを使用
from io import StringIO

buffer = StringIO()
for item in items:
    buffer.write(str(item))
result = buffer.getvalue()

Pythonツールとの統合

基本的なコマンド

# コードフォーマット
black .
isort .

# リンティング
ruff check .

# 型チェック
mypy .

# テスト
pytest

# カバレッジ
pytest --cov=src

CI/CD設定

# .github/workflows/checks.yml の例
name: Python Checks
on: [push, pull_request]

jobs:
  check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install black isort ruff mypy pytest
      - run: black --check .
      - run: isort --check-only .
      - run: ruff check .
      - run: mypy .
      - run: pytest

設計パターン

シングルトンパターン

# Good: メタクラスを使用したスレッドセーフなシングルトン
import threading

class SingletonMeta(type):
    """メタクラスによるスレッドセーフなシングルトン実装。"""
    _instances = {}
    _lock: threading.Lock = threading.Lock()

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            with cls._lock:
                if cls not in cls._instances:
                    instance = super().__call__(*args, **kwargs)
                    cls._instances[cls] = instance
        return cls._instances[cls]

class Database(metaclass=SingletonMeta):
    """アプリケーション全体で1つのデータベース接続。"""
    def __init__(self):
        self.connection = self._create_connection()

    def _create_connection(self):
        return "database_connection"

ファクトリーパターン

from abc import ABC, abstractmethod

class Document(ABC):
    @abstractmethod
    def render(self) -> str: ...

class PDFDocument(Document):
    def render(self) -> str:
        return "Rendering PDF document"

class WordDocument(Document):
    def render(self) -> str:
        return "Rendering Word document"

def document_factory(doc_type: str) -> Document:
    """ドキュメントタイプに基づいて適切なドキュメントを生成。"""
    factories = {
        "pdf": PDFDocument,
        "docx": WordDocument,
    }
    if doc_type not in factories:
        raise ValueError(f"Unknown document type: {doc_type}")
    return factories[doc_type]()

オブザーバーパターン

from typing import Protocol, List

class Observer(Protocol):
    def update(self, event: str, data: Any) -> None: ...

class Subject:
    """状態変化を監視し、オブザーバーに通知。"""
    def __init__(self):
        self._observers: List[Observer] = []

    def attach(self, observer: Observer) -> None:
        self._observers.append(observer)

    def detach(self, observer: Observer) -> None:
        self._observers.remove(observer)

    def notify(self, event: str, data: Any = None) -> None:
        for observer in self._observers:
            observer.update(event, data)

class Logger(Observer):
    def update(self, event: str, data: Any) -> None:
        print(f"Log: {event} - {data}")

class EmailNotifier(Observer):
    def update(self, event: str, data: Any) -> None:
        print(f"Email sent for: {event}")

テストパターン

pytestフィクスチャ

import pytest
from mypackage.models import User

@pytest.fixture
def sample_user() -> User:
    """テスト用のサンプルユーザーを作成。"""
    return User(
        id="test-123",
        name="Test User",
        email="[email protected]"
    )

@pytest.fixture
def user_repository():
    """インメモリユーザーリポジトリを作成。"""
    repo = InMemoryUserRepository()
    repo.add(User(id="1", name="Alice", email="[email protected]"))
    repo.add(User(id="2", name="Bob", email="[email protected]"))
    return repo

def test_get_user_by_id(user_repository):
    user = user_repository.get_by_id("1")
    assert user.name == "Alice"

def test_get_nonexistent_user(user_repository):
    with pytest.raises(NotFoundError):
        user_repository.get_by_id("999")

モックとパッチ

from unittest.mock import Mock, patch
import pytest

def test_fetch_user_data():
    """外部API呼び出しをモックしてテスト。"""
    mock_response = Mock()
    mock_response.json.return_value = {"id": "1", "name": "Alice"}
    mock_response.status_code = 200

    with patch("requests.get", return_value=mock_response):
        result = fetch_user_data("1")

    assert result["name"] == "Alice"

パラメータ化テスト

@pytest.mark.parametrize("input_data,expected", [
    ("()", True),
    ("()[]{}", True),
    ("(]", False),
    ("([)]", False),
    ("{[]}", True),
])
def test_valid_parentheses(input_data: str, expected: bool):
    """有効な括弧の組み合わせをテスト。"""
    assert is_valid(input_data) == expected

ログ記録パターン

構造化ログ

import logging
import json

class StructuredFormatter(logging.Formatter):
    """構造化されたJSONログを出力。"""
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "module": record.module,
            "function": record.funcName,
            "message": record.getMessage(),
        }
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_entry)

# 設定
handler = logging.StreamHandler()
handler.setFormatter(StructuredFormatter())
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

コンテキスト付きログ

import logging
from contextvars import ContextVar

request_id: ContextVar[str] = ContextVar('request_id', default='unknown')

class ContextFilter(logging.Filter):
    """ログレコードにコンテキスト情報を追加。"""
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()
        return True

# 設定
logger = logging.getLogger(__name__)
logger.addFilter(ContextFilter())

# リクエスト処理中に使用
async def handle_request(request):
    token = request_id.set(request.id)
    try:
        logger.info("Processing request")
        # リクエスト処理...
    finally:
        request_id.reset(token)

セキュリティパターン

入力バリデーション

from pydantic import BaseModel, EmailStr, Field, validator

class UserCreate(BaseModel):
    """ユーザー作成時の入力バリデーション。"""
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)

    @validator('username')
    def username_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError('Username must be alphanumeric')
        return v

安全なデータベースクエリ

# Bad: SQLインジェクションに対して脆弱
query = f"SELECT * FROM users WHERE name = '{user_input}'"

# Good: パラメータ化クエリを使用
query = "SELECT * FROM users WHERE name = ?"
cursor.execute(query, (user_input,))

環境設定

pyproject.toml 設定例

[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "mypackage"
version = "1.0.0"
description = "A sample Python package"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
    "requests>=2.28.0",
    "pydantic>=2.0.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.4.0",
    "black>=23.0.0",
    "isort>=5.12.0",
    "ruff>=0.1.0",
    "mypy>=1.5.0",
]

[tool.black]
line-length = 100
target-version = ["py310"]

[tool.isort]
profile = "black"
line_length = 100

[tool.ruff]
line-length = 100
target-version = "py310"

[tool.mypy]
python_version = "3.10"
strict = true

重要な注意点

• PEP 8に従う: Pythonコードの公式スタイルガイドです。
• 型ヒントを使用する: バグを早期に発見し、コードの明確性を向上させます。
• コンテキストマネージャーを採用する: 適切なリソース管理のために。
• ジェネレーターを活用する: 大規模データセットのメモリ効率を向上させます。
• 適切な抽象化レベルを選択する: やりすぎず、しかし十分すぎず。
• テストを書く: pytest、モック、フィクスチャを使用します。
• プロジェクトを整理する: 標準的なパッケージ構成に従います。
• コードレビューを実施する: パターンとベストプラクティスに関する一貫性を確保します。