Pythonベストプラクティス＆上級パターンガイド — プロのようにPythonを書く

my-project/
  src/
    my_package/
      __init__.py
      core.py
      utils.py
  tests/
    test_core.py
    test_utils.py
  pyproject.toml
  README.md

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "my-package"
version = "0.1.0"
description = "My awesome package"
requires-python = ">=3.11"
dependencies = [
    "httpx>=0.27",
    "pydantic>=2.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "mypy>=1.8",
    "ruff>=0.3",
]

[tool.ruff]
line-length = 88
target-version = "py311"

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

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

uv venv
uv pip install -e ".[dev]"
# またはlockfileベースのインストール
uv sync

# Python 3.10+ - typingインポート不要
def greet(name: str) -> str:
    return f"Hello, {name}!"

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

# Union型は | 演算子で表現
def parse_value(value: str | int | None) -> str:
    if value is None:
        return "empty"
    return str(value)

from typing import TypeVar, Generic

T = TypeVar("T")

class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        if not self._items:
            raise IndexError("Stack is empty")
        return self._items.pop()

    def peek(self) -> T:
        if not self._items:
            raise IndexError("Stack is empty")
        return self._items[-1]

# 使用例
int_stack: Stack[int] = Stack()
int_stack.push(42)
str_stack: Stack[str] = Stack()
str_stack.push("hello")

from typing import Protocol, runtime_checkable

@runtime_checkable
class Drawable(Protocol):
    def draw(self) -> str: ...

class Circle:
    def draw(self) -> str:
        return "Drawing circle"

class Square:
    def draw(self) -> str:
        return "Drawing square"

def render(shape: Drawable) -> None:
    print(shape.draw())

# CircleはDrawableを継承していないが、draw()があるため互換性あり
render(Circle())  # OK
render(Square())  # OK

[tool.mypy]
strict = true
python_version = "3.11"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true

[[tool.mypy.overrides]]
module = "tests.*"
disallow_untyped_defs = false

# プロジェクト全体をチェック
mypy src/
# 特定ファイルをチェック
mypy src/my_package/core.py

import functools
import time
from typing import Callable, ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")

def timer(func: Callable[P, R]) -> Callable[P, R]:
    """関数の実行時間を計測するデコレータ"""
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        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(n: int) -> int:
    """nまでの合計"""
    return sum(range(n))

slow_function(1_000_000)
# 出力: slow_function took 0.0312s

def retry(max_attempts: int = 3, delay: float = 1.0):
    """失敗時にリトライするデコレータ"""
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            last_exception: Exception | None = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    print(f"Attempt {attempt} failed: {e}")
                    if attempt < max_attempts:
                        time.sleep(delay)
            raise last_exception  # type: ignore
        return wrapper
    return decorator

@retry(max_attempts=5, delay=2.0)
def fetch_data(url: str) -> dict:
    # ネットワークリクエストのロジック
    ...

class CacheResult:
    """結果をキャッシュするクラスデコレータ"""
    def __init__(self, ttl_seconds: int = 300):
        self.ttl = ttl_seconds
        self.cache: dict[str, tuple[float, object]] = {}

    def __call__(self, func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            key = str(args) + str(kwargs)
            now = time.time()
            if key in self.cache:
                cached_time, cached_result = self.cache[key]
                if now - cached_time < self.ttl:
                    return cached_result  # type: ignore
            result = func(*args, **kwargs)
            self.cache[key] = (now, result)
            return result
        return wrapper

@CacheResult(ttl_seconds=60)
def expensive_computation(x: int) -> int:
    time.sleep(2)  # 高コスト処理のシミュレーション
    return x ** 2

class DatabaseConnection:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.connection = None

    def __enter__(self):
        print(f"Connecting to {self.connection_string}")
        self.connection = self._connect()
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.connection:
            self.connection.close()
            print("Connection closed")
        # Falseを返すと例外が伝播される
        return False

    def _connect(self):
        # 実際の接続ロジック
        ...

with DatabaseConnection("postgresql://localhost/mydb") as conn:
    conn.execute("SELECT 1")
# ブロックを出ると自動的に接続が閉じられる

from contextlib import contextmanager
import os

@contextmanager
def temporary_directory(path: str):
    """一時ディレクトリを作成し、使用後に削除"""
    os.makedirs(path, exist_ok=True)
    try:
        yield path
    finally:
        import shutil
        shutil.rmtree(path)

with temporary_directory("/tmp/work") as tmpdir:
    # tmpdirを使用した作業
    print(f"Working in {tmpdir}")
# ブロック終了時にディレクトリが自動削除

from contextlib import asynccontextmanager
import aiohttp

@asynccontextmanager
async def http_session():
    """aiohttpセッションのライフサイクル管理"""
    session = aiohttp.ClientSession()
    try:
        yield session
    finally:
        await session.close()

async def fetch_url(url: str) -> str:
    async with http_session() as session:
        async with session.get(url) as response:
            return await response.text()

# リスト: 全データをメモリにロード
numbers_list = [x ** 2 for x in range(10_000_000)]  # ~80MB

# ジェネレータ: 一つずつ生成、ほぼメモリ使用なし
numbers_gen = (x ** 2 for x in range(10_000_000))  # ~120B

def read_large_file(filepath: str, chunk_size: int = 8192):
    """大容量ファイルをチャンク単位で読み取り"""
    with open(filepath, "r") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

# 数GBのファイルもメモリの心配なく処理
for chunk in read_large_file("huge_log.txt"):
    process(chunk)

def flatten(nested: list) -> list:
    """ネストされたリストをフラット化"""
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)
        else:
            yield item

data = [1, [2, 3], [4, [5, 6]], 7]
print(list(flatten(data)))
# 出力: [1, 2, 3, 4, 5, 6, 7]

def chain(*iterables):
    """複数のイテラブルを一つに連結"""
    for iterable in iterables:
        yield from iterable

result = list(chain([1, 2], [3, 4], [5, 6]))
# 出力: [1, 2, 3, 4, 5, 6]

import csv
from typing import Iterator

def read_csv_rows(filename: str) -> Iterator[dict]:
    """CSVファイルを行ごとに読み取り"""
    with open(filename) as f:
        reader = csv.DictReader(f)
        yield from reader

def filter_active(rows: Iterator[dict]) -> Iterator[dict]:
    """アクティブユーザーのみフィルタリング"""
    for row in rows:
        if row["status"] == "active":
            yield row

def extract_emails(rows: Iterator[dict]) -> Iterator[str]:
    """メールフィールドを抽出"""
    for row in rows:
        yield row["email"]

# パイプライン: ファイル -> フィルタ -> 変換
pipeline = extract_emails(filter_active(read_csv_rows("users.csv")))
for email in pipeline:
    send_newsletter(email)

import asyncio

async def fetch_data(url: str, delay: float) -> str:
    print(f"Fetching {url}...")
    await asyncio.sleep(delay)  # ネットワークリクエストのシミュレーション
    return f"Data from {url}"

async def main():
    # 逐次実行: 3秒
    result1 = await fetch_data("api/users", 1.0)
    result2 = await fetch_data("api/posts", 1.0)
    result3 = await fetch_data("api/comments", 1.0)

    # 並列実行: 1秒
    results = await asyncio.gather(
        fetch_data("api/users", 1.0),
        fetch_data("api/posts", 1.0),
        fetch_data("api/comments", 1.0),
    )
    print(results)

asyncio.run(main())

import aiohttp
import asyncio

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return await response.json()

async def fetch_all_users(user_ids: list[int]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_url(session, f"https://api.example.com/users/{uid}")
            for uid in user_ids
        ]
        return await asyncio.gather(*tasks)

# 100人のユーザー情報を並列で取得
users = asyncio.run(fetch_all_users(list(range(1, 101))))

async def rate_limited_fetch(
    urls: list[str],
    max_concurrent: int = 10,
) -> list[str]:
    """同時リクエスト数を制限する関数"""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_with_limit(session: aiohttp.ClientSession, url: str) -> str:
        async with semaphore:
            async with session.get(url) as resp:
                return await resp.text()

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("api/users", 1.0))
        task2 = tg.create_task(fetch_data("api/posts", 1.0))
        task3 = tg.create_task(fetch_data("api/comments", 1.0))

    # すべてのタスクが完了した時点でここに到達
    print(task1.result(), task2.result(), task3.result())

from dataclasses import dataclass, field

@dataclass
class User:
    name: str
    email: str
    age: int
    tags: list[str] = field(default_factory=list)
    is_active: bool = True

    def display_name(self) -> str:
        return f"{self.name} ({self.email})"

user = User(name="Kim", email="kim@example.com", age=30)
print(user)
# User(name='Kim', email='kim@example.com', age=30, tags=[], is_active=True)

@dataclass(frozen=True)
class Point:
    x: float
    y: float

p = Point(1.0, 2.0)
# p.x = 3.0  # FrozenInstanceErrorが発生！

from typing import NamedTuple

class Coordinate(NamedTuple):
    latitude: float
    longitude: float
    altitude: float = 0.0

coord = Coordinate(37.5665, 126.9780)
lat, lng, alt = coord  # アンパック可能
print(coord.latitude)  # 属性アクセス

from pydantic import BaseModel, EmailStr, Field, field_validator

class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(ge=0, le=150)
    tags: list[str] = []

    @field_validator("name")
    @classmethod
    def name_must_be_capitalized(cls, v: str) -> str:
        if not v[0].isupper():
            raise ValueError("Name must start with uppercase")
        return v

# 正しい入力
user = UserCreate(name="Kim", email="kim@example.com", age=30)
print(user.model_dump_json())

# 不正な入力 -> ValidationError
try:
    bad_user = UserCreate(name="", email="not-an-email", age=-5)
except Exception as e:
    print(e)

class DatabasePool:
    _instance: "DatabasePool | None" = None

    def __new__(cls) -> "DatabasePool":
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._initialize()
        return cls._instance

    def _initialize(self) -> None:
        self.connections: list = []
        print("Pool initialized")

# 常に同じインスタンスを返す
pool1 = DatabasePool()
pool2 = DatabasePool()
assert pool1 is pool2

# db.py
_pool = None

def get_pool():
    global _pool
    if _pool is None:
        _pool = create_pool()
    return _pool

from abc import ABC, abstractmethod

class Notification(ABC):
    @abstractmethod
    def send(self, message: str) -> None: ...

class EmailNotification(Notification):
    def send(self, message: str) -> None:
        print(f"Email: {message}")

class SlackNotification(Notification):
    def send(self, message: str) -> None:
        print(f"Slack: {message}")

class SMSNotification(Notification):
    def send(self, message: str) -> None:
        print(f"SMS: {message}")

def create_notification(channel: str) -> Notification:
    factories: dict[str, type[Notification]] = {
        "email": EmailNotification,
        "slack": SlackNotification,
        "sms": SMSNotification,
    }
    if channel not in factories:
        raise ValueError(f"Unknown channel: {channel}")
    return factories[channel]()

notif = create_notification("slack")
notif.send("Hello!")

from typing import Callable

class EventEmitter:
    def __init__(self) -> None:
        self._listeners: dict[str, list[Callable]] = {}

    def on(self, event: str, callback: Callable) -> None:
        self._listeners.setdefault(event, []).append(callback)

    def emit(self, event: str, *args, **kwargs) -> None:
        for callback in self._listeners.get(event, []):
            callback(*args, **kwargs)

# 使用例
emitter = EventEmitter()
emitter.on("user_created", lambda user: print(f"Welcome, {user}!"))
emitter.on("user_created", lambda user: send_email(user))
emitter.emit("user_created", "Kim")

from typing import Protocol

class SortStrategy(Protocol):
    def sort(self, data: list[int]) -> list[int]: ...

class BubbleSort:
    def sort(self, data: list[int]) -> list[int]:
        arr = data.copy()
        n = len(arr)
        for i in range(n):
            for j in range(0, n - i - 1):
                if arr[j] > arr[j + 1]:
                    arr[j], arr[j + 1] = arr[j + 1], arr[j]
        return arr

class QuickSort:
    def sort(self, data: list[int]) -> list[int]:
        if len(data) <= 1:
            return data
        pivot = data[len(data) // 2]
        left = [x for x in data if x < pivot]
        middle = [x for x in data if x == pivot]
        right = [x for x in data if x > pivot]
        return self.sort(left) + middle + self.sort(right)

class Sorter:
    def __init__(self, strategy: SortStrategy) -> None:
        self._strategy = strategy

    def sort(self, data: list[int]) -> list[int]:
        return self._strategy.sort(data)

# 実行時にストラテジーを切り替え
sorter = Sorter(QuickSort())
print(sorter.sort([3, 1, 4, 1, 5, 9]))

import cProfile
import pstats

def expensive_function():
    total = 0
    for i in range(1_000_000):
        total += i ** 2
    return total

# プロファイリング実行
profiler = cProfile.Profile()
profiler.enable()
expensive_function()
profiler.disable()

# 結果出力
stats = pstats.Stats(profiler)
stats.sort_stats("cumulative")
stats.print_stats(10)

pip install line_profiler

@profile  # line_profilerデコレータ
def process_data(data: list[int]) -> list[int]:
    result = []                    # ほぼ0
    for item in data:              # ループオーバーヘッド
        if item % 2 == 0:         # 条件チェック
            result.append(item * 2)  # 実際の処理
    return result

from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# キャッシュなし: O(2^n) -> 非常に遅い
# キャッシュあり: O(n) -> 即座に完了
print(fibonacci(100))

# キャッシュ統計を確認
print(fibonacci.cache_info())
# CacheInfo(hits=98, misses=101, maxsize=128, currsize=101)

import timeit

# forループ - 遅い
def squares_loop(n: int) -> list[int]:
    result = []
    for i in range(n):
        result.append(i ** 2)
    return result

# リスト内包表記 - 速い（Cレベルの最適化）
def squares_comp(n: int) -> list[int]:
    return [i ** 2 for i in range(n)]

# ベンチマーク
n = 100_000
t_loop = timeit.timeit(lambda: squares_loop(n), number=100)
t_comp = timeit.timeit(lambda: squares_comp(n), number=100)
print(f"Loop: {t_loop:.3f}s, Comprehension: {t_comp:.3f}s")
# 内包表記は通常20〜30%高速

# 1. 文字列結合にはjoin()を使用
words = ["hello", "world", "python"]
result = " ".join(words)  # "+" 演算子よりはるかに高速

# 2. メンバーシップテストにはsetを使用
valid_ids = set(range(10000))
if 42 in valid_ids:  # O(1) - リストのO(n)より高速
    pass

# 3. collectionsの活用
from collections import Counter, defaultdict

# 頻度カウント
word_count = Counter(["apple", "banana", "apple", "cherry", "apple"])
print(word_count.most_common(2))  # [('apple', 3), ('banana', 1)]

# デフォルト辞書
grouped = defaultdict(list)
for item in [("A", 1), ("B", 2), ("A", 3)]:
    grouped[item[0]].append(item[1])

# tests/test_calculator.py
def add(a: int, b: int) -> int:
    return a + b

def test_add_positive():
    assert add(2, 3) == 5

def test_add_negative():
    assert add(-1, -1) == -2

def test_add_zero():
    assert add(0, 0) == 0

# テスト実行
pytest tests/ -v
# カバレッジ付き
pytest --cov=src --cov-report=term-missing

import pytest

@pytest.fixture
def sample_users() -> list[dict]:
    return [
        {"name": "Kim", "age": 30},
        {"name": "Lee", "age": 25},
        {"name": "Park", "age": 35},
    ]

@pytest.fixture
def db_connection():
    conn = create_connection()
    yield conn  # テスト実行
    conn.close()  # クリーンアップ

def test_user_count(sample_users):
    assert len(sample_users) == 3

def test_oldest_user(sample_users):
    oldest = max(sample_users, key=lambda u: u["age"])
    assert oldest["name"] == "Park"

@pytest.mark.parametrize("input_val, expected", [
    ("hello", 5),
    ("", 0),
    ("python", 6),
    ("   ", 3),
])
def test_string_length(input_val: str, expected: int):
    assert len(input_val) == expected

@pytest.mark.parametrize("a, b, expected", [
    (2, 3, 5),
    (-1, 1, 0),
    (0, 0, 0),
    (100, 200, 300),
])
def test_add(a: int, b: int, expected: int):
    assert add(a, b) == expected

from unittest.mock import Mock, patch, AsyncMock

class UserService:
    def __init__(self, api_client):
        self.api = api_client

    def get_user(self, user_id: int) -> dict:
        response = self.api.get(f"/users/{user_id}")
        return response.json()

def test_get_user():
    # MockのAPIクライアントを作成
    mock_api = Mock()
    mock_api.get.return_value.json.return_value = {
        "id": 1,
        "name": "Kim",
    }

    service = UserService(mock_api)
    user = service.get_user(1)

    assert user["name"] == "Kim"
    mock_api.get.assert_called_once_with("/users/1")

# patchによるモック
@patch("my_module.requests.get")
def test_fetch_data(mock_get):
    mock_get.return_value.status_code = 200
    mock_get.return_value.json.return_value = {"data": "test"}

    result = fetch_data("https://api.example.com")
    assert result["data"] == "test"

import pytest

@pytest.mark.asyncio
async def test_async_fetch():
    result = await fetch_data("https://api.example.com/users/1")
    assert "name" in result

@pytest.mark.asyncio
async def test_concurrent_requests():
    results = await asyncio.gather(
        fetch_data("url1"),
        fetch_data("url2"),
    )
    assert len(results) == 2