python-patterns

Python 開發模式

用於建立穩健、高效且易於維護應用程式的慣用 Python 模式與最佳實踐。

何時啟動

• 編寫新的 Python 程式碼
• 審查 Python 程式碼
• 重構現有 Python 程式碼
• 設計 Python 套件/模組

核心原則

1. 可讀性至上

Python 優先考慮可讀性。程式碼應清晰易懂。

# 好：清晰且易於閱讀
def get_active_users(users: list[User]) -> list[User]:
    """從提供的清單中僅回傳活躍使用者。"""
    return [user for user in users if user.is_active]


# 差：聰明但令人困惑
def get_active_users(u):
    return [x for x in u if x.a]

2. 顯式優於隱式

避免神奇魔法；清楚說明程式碼的用途。

# 好：顯式設定
import logging

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

# 差：隱藏的副作用
import some_module
some_module.setup()  # 這在做什麼？

3. EAFP - 請求原諒比請求許可更容易

Python 偏好使用例外處理而非預先檢查條件。

# 好：EAFP 風格
def get_value(dictionary: dict, key: str, default_value: Any = None) -> Any:
    try:
        return dictionary[key]
    except KeyError:
        return default_value

# 差：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]:
    """處理使用者並回傳更新後的 User 或 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:
    """回傳第一個項目，如果清單為空則回傳 None。"""
    return items[0] if items else None

基於協定的鴨子型別

from typing import Protocol

class Renderable(Protocol):
    def render(self) -> str:
        """將物件渲染為字串。"""

class Image:
    def render(self) -> str:
        return "<image>"

def render_all(items: list[Renderable]) -> str:
    """渲染所有實作 Renderable 協定的項目。"""
    return "\n".join(item.render() for item in items)

render_all([Image()])  # 輸出: <image>

錯誤處理模式

特定例外處理

# 好：捕捉特定例外
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"設定檔未找到: {path}") from e
    except json.JSONDecodeError as e:
        raise ConfigError(f"設定檔中的 JSON 無效: {path}") from e

# 差：裸 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"無法解析資料: {data}") from e

自訂例外階層

class AppError(Exception):
    """所有應用程式錯誤的基礎例外。"""
    pass

class ValidationError(AppError):
    """當輸入驗證失敗時引發。"""
    pass

class NotFoundError(AppError):
    """當要求的資源未找到時引發。"""
    pass

# 使用方式
def get_user(user_id: str) -> User:
    user = db.find_user(user_id)
    if not user:
        raise NotFoundError(f"使用者未找到: {user_id}")
    return user

情境管理器

資源管理

# 好：使用情境管理器
def process_file(path: str) -> str:
    with open(path, 'r') as f:
        return f.read()

# 差：手動資源管理
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):
    """用於計時一段程式碼的情境管理器。"""
    start = time.perf_counter()
    yield
    elapsed = time.perf_counter() - start
    print(f"{name} 花了 {elapsed:.4f} 秒")

# 使用方式
with timer("資料處理"):
    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)

推導式與生成器

列表推導式

# 好：用列表推導式進行簡單轉換
names = [user.name for user in users if user.is_active]

# 差：手動迴圈
names = []
for user in users:
    if user.is_active:
        names.append(user.name)

# 複雜的推導式應展開
# 差：過於複雜
result = [x * 2 for x in items if x > 0 if x % 2 == 0]

# 好：使用生成器函式
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

生成器表達式

# 好：用生成器進行惰性求值
total = sum(x * x for x in range(1_000_000))

# 差：建立大型中間列表
total = sum([x * x for x in range(1_000_000)])

生成器函式

def read_large_file(path: str) -> Iterator[str]:
    """逐行讀取大型檔案。"""
    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:
    """自動具有 __init__、__repr__ 和 __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"無效的電子郵件: {self.email}")
        # 驗證年齡範圍
        if self.age < 0 or self.age > 150:
            raise ValueError(f"無效的年齡: {self.age}")

具名元組

from typing import NamedTuple

class Point(NamedTuple):
    """不可變的 2D 點。"""
    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:
    """用於計時函式執行的裝飾器。"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} 花了 {elapsed:.4f}s")
        return result
    return wrapper

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

# slow_function() 輸出: slow_function 花了 1.0012s

參數化裝飾器

def repeat(times: int):
    """重複執行函式多次的裝飾器。"""
    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") 回傳 ["Hello, Alice!", "Hello, Alice!", "Hello, Alice!"]

基於類別的裝飾器

class CountCalls:
    """計算函式被呼叫次數的裝飾器。"""
    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__} 已被呼叫 {self.count} 次")
        return self.func(*args, **kwargs)

@CountCalls
def process():
    pass

# 每次呼叫 process() 都會印出呼叫次數

並行模式

I/O 密集型任務的執行緒

import concurrent.futures
import threading

def fetch_url(url: str) -> str:
    """獲取 URL（I/O 密集型操作）。"""
    import urllib.request
    with urllib.request.urlopen(url) as response:
        return response.read().decode()

def fetch_all_urls(urls: list[str]) -> dict[str, str]:
    """使用執行緒同時獲取多個 URL。"""
    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"錯誤: {e}"
    return results

CPU 密集型任務的多程序

def process_data(data: list[int]) -> int:
    """CPU 密集型計算。"""
    return sum(x ** 2 for x in data)

def process_all(datasets: list[list[int]]) -> list[int]:
    """使用多個程序處理多個資料集。"""
    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:
    """非同步獲取 URL。"""
    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]:
    """同時獲取多個 URL。"""
    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

Import 慣例

# 好：Import 順序 - 標準函式庫、第三方、本地
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

# 好：使用 isort 進行自動 import 排序
# pip install isort

init.py 用於套件匯出

# mypackage/__init__.py
"""mypackage - 一個範例 Python 套件。"""

__version__ = "1.0.0"

# 在套件層級匯出主要類別/函式
from mypackage.models import User, Post
from mypackage.utils import format_name

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

記憶體與效能

使用 slots 提高記憶體效率

# 差：一般類別使用 __dict__（更多記憶體）
class Point:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

# 好：__slots__ 減少記憶體使用
class Point:
    __slots__ = ['x', 'y']

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

使用生成器處理大型資料

# 差：在記憶體中回傳完整列表
def read_lines(path: str) -> list[str]:
    with open(path) as f:
        return [line.strip() for line in f]

# 好：逐行產生
from typing import Iterator

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

避免在迴圈中進行字串拼接

# 差：由於字串不可變性，導致 O(n²)
result = ""
for item in items:
    result += str(item)

# 好：使用 join 達到 O(n)
result = "".join(str(item) for item in items)

# 好：使用 StringIO 建構
from io import StringIO

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

Python 工具整合

基本指令

# 程式碼格式化
black .
isort .

# Linting
ruff check .

# 型別檢查
mypy .

# 測試
pytest -v
pytest --cov=mypackage tests/

# 套件建置
python -m build

pyproject.toml 範例

[build-system]
requires = ["setuptools>=64", "wheel"]
build-backend = "setuptools.backends._legacy:_Backend"

[project]
name = "mypackage"
version = "1.0.0"
description = "A sample Python package"
requires-python = ">=3.9"
dependencies = []

[project.optional-dependencies]
dev = [
    "pytest>=7",
    "pytest-cov>=4",
    "black>=23",
    "isort>=5",
    "ruff>=0.1",
    "mypy>=1",
]

[tool.black]
line-length = 88
target-version = ["py39"]

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

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

[tool.mypy]
python_version = "3.9"
warn_return_any = true
warn_unused_configs = true

[tool.pytest.ini_options]
addopts = "-v"
testpaths = ["tests"]

安全性模式

輸入驗證

# 好：使用 Pydantic 或資料類別進行驗證
from pydantic import BaseModel, EmailStr, Field

class UserInput(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)

# 使用方式
try:
    user = UserInput(name="Alice", email="[email protected]", age=30)
except ValidationError as e:
    print(f"驗證失敗: {e}")

# 差：未經驗證的輸入
def create_user(name: str, email: str, age: int):
    # 如果輸入無效，這裡可能導致問題
    pass

避免命令注入

import subprocess

# 好：使用子程序列出目錄內容
subprocess.run(["ls", "-l", "/safe/path"], check=True)

# 差：將未經驗證的輸入傳遞給 shell
import os
user_input = "some; rm -rf /"  # 惡意輸入
os.system(f"ls {user_input}")  # 危險！

# 使用 shlex 進行安全引數處理
import shlex
safe_cmd = shlex.split(f"ls {user_input}")
subprocess.run(safe_cmd, check=True)  # 引發 FileNotFoundError，但不會執行惡意程式碼

安全密碼處理

import hashlib
import secrets

# 好：使用安全的密碼雜湊
from passlib.hash import bcrypt

def hash_password(password: str) -> str:
    """使用 bcrypt 對密碼進行雜湊處理。"""
    return bcrypt.hash(password)

def verify_password(password: str, hashed: str) -> bool:
    """驗證密碼。"""
    return bcrypt.verify(password, hashed)

# 差：使用基本 SHA256
import hashlib
def hash_password(password: str) -> str:
    return hashlib.sha256(password.encode()).hexdigest()
# 這容易受到彩虹表攻擊！

測試模式

pytest 基礎

# conftest.py
import pytest
from mypackage import create_app

@pytest.fixture
def app():
    app = create_app({"TESTING": True})
    yield app

@pytest.fixture
def client(app):
    return app.test_client()

# test_api.py
def test_health_endpoint(client):
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json == {"status": "healthy"}

def test_user_creation(client):
    response = client.post("/users", json={
        "name": "Alice",
        "email": "[email protected]"
    })
    assert response.status_code == 201
    assert "id" in response.json

模擬 external services

from unittest.mock import Mock, patch

@patch("mypackage.external_service.api_call")
def test_data_fetching(mock_api):
    # 設定模擬回傳值
    mock_api.return_value = {"data": "test"}
    
    # 測試函式
    result = fetch_and_process_data()
    
    # 驗證結果
    assert result == expected_result
    mock_api.assert_called_once()

基於屬性的測試

from hypothesis import given, strategies as st

def test_list_reversal():
    """測試列表反轉正確性。"""
    assert reverse_list([1, 2, 3]) == [3, 2, 1]
    assert reverse_list([]) == []
    assert reverse_list([1]) == [1]

@given(st.lists(st.integers()))
def test_reversal_properties(lst):
    """基於屬性的測試：反轉兩次應得到原始列表。"""
    assert reverse_list(reverse_list(lst)) == lst

錯誤處理與日誌記錄

結構化日誌記錄

import logging
from logging.handlers import RotatingFileHandler
import json

class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if hasattr(record, "extra"):
            log_entry.update(record.extra)
        return json.dumps(log_entry)

# 設定
logger = logging.getLogger("myapp")
handler = RotatingFileHandler("app.log", maxBytes=10_000_000, backupCount=5)
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# 使用方式
logger.info("使用者已建立", extra={"user_id": "123", "action": "create"})

優雅降級

import functools
from typing import Callable

def fallback_on_error(default_value: Any = None):
    """裝飾器：如果被裝飾的函式引發例外，則回傳預設值。"""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.error(f"{func.__name__} 失敗: {e}")
                return default_value
        return wrapper
    return decorator

@fallback_on_error(default_value={})
def fetch_user_data(user_id: str) -> dict:
    """從外部服務獲取使用者資料。"""
    response = requests.get(f"https://api.example.com/users/{user_id}")
    response.raise_for_status()
    return response.json()

結論

遵循這些 Python 模式可確保你的程式碼是：

• 可讀的：易於理解並遵循 PEP 8
• 可維護的：使用型別提示與適當的抽象
• 高效能的：使用惰性求值與記憶體最佳化
• 安全的：遵循輸入驗證與密碼處理最佳實踐
• 經過充分測試的：使用 pytest 與模擬技術

記住：Python 的禪宗是：「簡單勝於複雜。」始終優先考慮簡潔明瞭的解決方案。