my_project/
├── src/                      # 源代碼目錄
│   └── my_package/           # 主包目錄
│       ├── __init__.py       # 包初始化文件
│       ├── core/             # 核心模塊
│       │   ├── __init__.py
│       │   ├── engine.py
│       │   └── utils.py
│       ├── plugins/          # 插件系統
│       │   ├── __init__.py
│       │   └── analysis.py
│       └── cli.py            # 命令行接口
├── tests/                    # 測試代碼
│   ├── unit/                 # 單元測試
│   └── integration/          # 集成測試
├── docs/                     # 文檔
├── scripts/                  # 實用腳本
├── pyproject.toml            # 項目配置
├── README.md
└── LICENSE

# 模塊化設計檢查清單
modularity_checklist = {
    '單一職責': '每個模塊/類只做一件事',
    '低耦合': '模塊間最小化依賴',
    '高內聚': '相關功能組織在一起',
    '明確接口': '模塊間通過定義良好的接口通信',
    '分層架構': '分離不同抽象層次(表現層/業務層/數據層)'
}

from abc import ABC, abstractmethod
from typing import Dict, Any

class DataProcessor(ABC):
    """數據處理抽象基類"""
    
    @abstractmethod
    def load(self, source: str) -> Any:
        """從源加載數據"""
        pass
    
    @abstractmethod
    def process(self, data: Any) -> Any:
        """處理數據"""
        pass
    
    @abstractmethod
    def save(self, data: Any, destination: str) -> bool:
        """保存處理結果"""
        pass

class CSVProcessor(DataProcessor):
    """CSV數據處理實現"""
    
    def load(self, source: str) -> list:
        import csv
        with open(source, 'r') as f:
            return list(csv.reader(f))
    
    def process(self, data: list) -> list:
        return [row for row in data if any(field.strip() for field in row)]
    
    def save(self, data: list, destination: str) -> bool:
        import csv
        with open(destination, 'w') as f:
            writer = csv.writer(f)
            writer.writerows(data)
        return True

from typing import Protocol, runtime_checkable

@runtime_checkable
class AnalyzerProtocol(Protocol):
    """分析器協議"""
    
    def analyze(self, text: str) -> dict:
        """分析文本"""
        ...
    
    @property
    def version(self) -> str:
        """分析器版本"""
        ...

class SimpleAnalyzer:
    """簡單分析器實現"""
    
    def analyze(self, text: str) -> dict:
        return {'length': len(text), 'words': len(text.split())}
    
    @property
    def version(self) -> str:
        return "1.0"

def use_analyzer(analyzer: AnalyzerProtocol):
    """使用符合協議的分析器"""
    print(f"Using analyzer v{analyzer.version}")
    result = analyzer.analyze("Hello world")
    print(result)

# 使用示例
analyzer = SimpleAnalyzer()
use_analyzer(analyzer)  # 類型檢查通過

from dataclasses import dataclass
from dependency_injector import containers, providers

@dataclass
class DatabaseConfig:
    host: str
    port: int
    user: str
    password: str

class DatabaseService:
    def __init__(self, config: DatabaseConfig):
        self.config = config
    
    def query(self, sql: str):
        # 實際數據庫操作
        print(f"Querying {self.config.host}: {sql}")
        return []

class CoreContainer(containers.DeclarativeContainer):
    """核心依賴容器"""
    
    config = providers.Configuration()
    
    db_config = providers.Singleton(
        DatabaseConfig,
        host=config.db.host,
        port=config.db.port,
        user=config.db.user,
        password=config.db.password
    )
    
    database = providers.Singleton(
        DatabaseService,
        config=db_config
    )

# 使用示例
container = CoreContainer()
container.config.from_dict({
    "db": {
        "host": "localhost",
        "port": 5432,
        "user": "admin",
        "password": "secret"
    }
})

db_service = container.database()
results = db_service.query("SELECT * FROM users")

from abc import ABC, abstractmethod

class NotificationService(ABC):
    """通知服務抽象"""
    
    @abstractmethod
    def send(self, message: str, recipient: str) -> bool:
        """發送通知"""
        pass

class EmailService(NotificationService):
    """郵件通知實現"""
    
    def send(self, message: str, recipient: str) -> bool:
        print(f"Sending email to {recipient}: {message}")
        return True

class UserManager:
    """用户管理器(依賴抽象而非具體實現)"""
    
    def __init__(self, notifier: NotificationService):
        self.notifier = notifier
    
    def register_user(self, username: str, email: str):
        # 註冊邏輯...
        self.notifier.send(
            "Welcome to our platform!", 
            email
        )

# 使用示例
email_service = EmailService()
user_manager = UserManager(email_service)
user_manager.register_user("johndoe", "john@example.com")

import os
from typing import Dict, Any
from pathlib import Path
import json
import yaml

class ConfigManager:
    """分層配置管理器"""
    
    def __init__(self):
        self._layers = {
            'defaults': {},
            'file': {},
            'env': {},
            'runtime': {}
        }
    
    def load_defaults(self, config: Dict[str, Any]):
        """加載默認配置"""
        self._layers['defaults'].update(config)
    
    def load_file(self, path: str):
        """從文件加載配置"""
        path = Path(path)
        if path.suffix == '.json':
            self._layers['file'].update(json.loads(path.read_text()))
        elif path.suffix in ('.yaml', '.yml'):
            self._layers['file'].update(yaml.safe_load(path.read_text()))
    
    def load_env(self, prefix: str = "APP_"):
        """從環境變量加載配置"""
        self._layers['env'].update({
            k[len(prefix):].lower(): v
            for k, v in os.environ.items()
            if k.startswith(prefix)
        })
    
    def get(self, key: str, default: Any = None) -> Any:
        """獲取配置值(按優先級)"""
        for layer in ('runtime', 'env', 'file', 'defaults'):
            if key in self._layers[layer]:
                return self._layers[layer][key]
        return default

# 使用示例
config = ConfigManager()
config.load_defaults({
    'database': {
        'host': 'localhost',
        'port': 3306
    }
})
config.load_file('config/production.yaml')
config.load_env('MY_APP_')

db_host = config.get('database.host')

from pydantic import BaseModel, validator
from typing import Literal

class DatabaseConfig(BaseModel):
    """數據庫配置模型"""
    
    engine: Literal['mysql', 'postgres', 'sqlite']
    host: str = 'localhost'
    port: int = 3306
    timeout: int = 30
    
    @validator('port')
    def validate_port(cls, v):
        if not 0 < v < 65535:
            raise ValueError('Port must be between 1-65534')
        return v

# 使用示例
try:
    config = DatabaseConfig(**raw_config)
except ValueError as e:
    print(f"Invalid config: {e}")

import importlib
import pkgutil
from pathlib import Path
from typing import Dict, Type

class PluginManager:
    """插件管理器"""
    
    def __init__(self):
        self._plugins: Dict[str, Type] = {}
    
    def discover(self, plugin_dir: str):
        """發現並加載插件"""
        plugin_path = Path(plugin_dir)
        for finder, name, _ in pkgutil.iter_modules([str(plugin_path)]):
            module = finder.find_module(name).load_module()
            if hasattr(module, 'register_plugin'):
                module.register_plugin(self)
    
    def register(self, name: str, plugin_class: Type):
        """註冊插件類"""
        self._plugins[name] = plugin_class
    
    def create_instance(self, name: str, *args, **kwargs):
        """創建插件實例"""
        if name not in self._plugins:
            raise ValueError(f"Unknown plugin: {name}")
        return self._plugins[name](*args, **kwargs)

# 插件示例 (plugins/analysis.py)
"""
class AnalysisPlugin:
    def process(self, data):
        return data * 2

def register_plugin(manager):
    manager.register('analysis', AnalysisPlugin)
"""

[project.entry-points."myapp.plugins"]
csv = "myapp.plugins.csv:register_plugin"
json = "myapp.plugins.json:register_plugin"

from importlib.metadata import entry_points

def load_plugins():
    """加載入口點插件"""
    manager = PluginManager()
    for ep in entry_points().get('myapp.plugins', []):
        register_func = ep.load()
        register_func(manager)
    return manager

from typing import Optional, Dict, Any
from dataclasses import dataclass

@dataclass
class AppError(Exception):
    """應用基礎錯誤類"""
    
    code: str
    message: str
    details: Optional[Dict[str, Any]] = None
    
    def __str__(self):
        return f"[{self.code}] {self.message}"

class ValidationError(AppError):
    """驗證錯誤"""
    
    def __init__(self, field: str, reason: str):
        super().__init__(
            code="VALIDATION_ERROR",
            message=f"Invalid field '{field}'",
            details={'field': field, 'reason': reason}
        )

def process_data(data: dict):
    """處理數據(帶有詳細錯誤信息)"""
    if 'username' not in data:
        raise ValidationError('username', 'required')
    if len(data.get('username', '')) < 3:
        raise ValidationError('username', 'too short')
    
    # 處理邏輯...
    return True

# 使用示例
try:
    process_data({'username': 'a'})
except AppError as e:
    print(f"Error: {e}\nDetails: {e.details}")

import logging
import json
from typing import Dict, Any

class JSONFormatter(logging.Formatter):
    """JSON日誌格式化器"""
    
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName
        }
        if hasattr(record, 'context'):
            log_data.update(record.context)
        return json.dumps(log_data)

def setup_logging():
    """配置結構化日誌"""
    logger = logging.getLogger('myapp')
    logger.setLevel(logging.INFO)
    
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    
    logger.addHandler(handler)
    return logger

# 使用示例
logger = setup_logging()
logger.info("Processing started", extra={
    'context': {
        'job_id': 123,
        'input_size': 1024
    }
})

# 測試類型分佈
test_pyramid = {
    '單元測試': {
        '比例': 70,
        '工具': ['pytest', 'unittest'],
        '特點': '快速執行，隔離測試'
    },
    '集成測試': {
        '比例': 20,
        '工具': ['pytest', 'docker'],
        '特點': '驗證模塊間交互'
    },
    '端到端測試': {
        '比例': 10,
        '工具': ['selenium', 'playwright'],
        '特點': '驗證完整流程'
    }
}

# 測試目錄結構示例
"""
tests/
├── unit/                 # 單元測試
│   ├── core/             # 核心模塊測試
│   └── utils/            # 工具類測試
├── integration/          # 集成測試
│   ├── database/         # 數據庫集成
│   └── api/              # API集成
└── e2e/                  # 端到端測試
    ├── web/              # Web界面測試
    └── cli/              # 命令行測試
"""

import pytest
from tempfile import TemporaryDirectory
from pathlib import Path

@pytest.fixture
def temp_config():
    """臨時配置文件夾具"""
    with TemporaryDirectory() as temp_dir:
        config_path = Path(temp_dir) / "config.yaml"
        config_path.write_text("""
        database:
          host: test.db
          port: 3306
        """)
        yield config_path

@pytest.fixture
def database_connection(temp_config):
    """數據庫連接夾具"""
    from myapp.database import create_connection
    conn = create_connection(temp_config)
    yield conn
    conn.close()

def test_database_operations(database_connection):
    """使用夾具的測試示例"""
    result = database_connection.query("SELECT 1")
    assert result == [1]

# 1. 插件系統架構決策

## 狀態
提議

## 背景
需要支持動態加載功能模塊

## 決策
採用基於入口點(entry_points)的插件系統

## 後果
- ✅ 靈活擴展功能
- ✅ 鬆耦合架構
- ❌ 增加初始化複雜度

```mermaid
graph TD
    A[客户端] --> B[API網關]
    B --> C{路由}
    C --> D[服務A]
    C --> E[服務B]
    
    D --> F[(數據庫A)]
    E --> G[(數據庫B)]
    
    style A fill:#f9f,stroke:#333
    style B fill:#bbf,stroke:#333
```