生产级代码实践¶

# 异常层次结构设计示例

from typing import Optional, Any
from datetime import datetime


class AppException(Exception):
    """应用基础异常类"""

    def __init__(self, message: str, code: str = "APP_ERROR", details: Optional[dict] = None):
        super().__init__(message)
        self.message = message
        self.code = code
        self.details = details or {}
        self.timestamp = datetime.now()

    def to_dict(self) -> dict:
        """转换为字典格式，便于日志记录和 API 返回"""
        return {
            "error": self.__class__.__name__,
            "message": self.message,
            "code": self.code,
            "details": self.details,
            "timestamp": self.timestamp.isoformat()
        }


class DataException(AppException):
    """数据相关异常基类"""
    pass


class DataNotFoundException(DataException):
    """数据未找到异常"""

    def __init__(self, resource: str, identifier: Any):
        super().__init__(
            message=f"Resource '{resource}' not found with identifier: {identifier}",
            code="DATA_NOT_FOUND",
            details={"resource": resource, "identifier": str(identifier)}
        )


class DataValidationException(DataException):
    """数据验证异常"""

    def __init__(self, field: str, value: Any, reason: str):
        super().__init__(
            message=f"Validation failed for field '{field}': {reason}",
            code="DATA_VALIDATION_ERROR",
            details={"field": field, "value": str(value), "reason": reason}
        )


class NetworkException(AppException):
    """网络相关异常基类"""
    pass


class NetworkTimeoutException(NetworkException):
    """网络超时异常"""

    def __init__(self, url: str, timeout: float):
        super().__init__(
            message=f"Request to {url} timed out after {timeout}s",
            code="NETWORK_TIMEOUT",
            details={"url": url, "timeout": timeout}
        )


class RateLimitException(NetworkException):
    """速率限制异常"""

    def __init__(self, service: str, retry_after: int):
        super().__init__(
            message=f"Rate limit exceeded for '{service}', retry after {retry_after}s",
            code="RATE_LIMIT_EXCEEDED",
            details={"service": service, "retry_after": retry_after}
        )


# 使用示例
if __name__ == "__main__":
    try:
        raise DataNotFoundException("user", "user_12345")
    except DataException as e:
        print(f"Caught data exception: {e.to_dict()}")

    try:
        raise DataValidationException("email", "invalid-email", "must contain @")
    except DataException as e:
        print(f"Caught validation error: {e}")

# 自定义异常类最佳实践

from typing import Any, Optional, List
import traceback


class RichException(Exception):
    """增强型异常类，提供更丰富的上下文信息"""

    def __init__(
        self,
        message: str,
        cause: Optional[Exception] = None,
        context: Optional[dict] = None,
        suggestions: Optional[List[str]] = None
    ):
        super().__init__(message)
        self.message = message
        self.cause = cause
        self.context = context or {}
        self.suggestions = suggestions or []
        self.stack_trace = traceback.format_exc()

    def __str__(self) -> str:
        parts = [self.message]
        if self.suggestions:
            parts.append("Suggestions: " + "; ".join(self.suggestions))
        return " | ".join(parts)

    def to_dict(self) -> dict:
        return {
            "message": self.message,
            "cause": str(self.cause) if self.cause else None,
            "context": self.context,
            "suggestions": self.suggestions,
            "stack_trace": self.stack_trace
        }


class ConfigException(RichException):
    """配置相关异常"""
    pass


# 异常工厂函数
class ExceptionFactory:
    """异常工厂，用于创建标准化的异常实例"""

    @staticmethod
    def not_found(resource: str, identifier: Any) -> Exception:
        return RichException(
            message=f"{resource} not found: {identifier}",
            suggestions=[
                f"Check if the {resource.lower()} exists",
                "Verify the identifier is correct"
            ]
        )

    @staticmethod
    def invalid_input(field: str, reason: str) -> Exception:
        return RichException(
            message=f"Invalid input for '{field}': {reason}",
            context={"field": field}
        )


# 使用示例
if __name__ == "__main__":
    try:
        raise ExceptionFactory.not_found("User", "user_999")
    except RichException as e:
        print(f"Exception: {e}")
        print(f"Details: {e.to_dict()}")

# 上下文管理器完整指南

import time
import sqlite3
import json
from typing import Any, Optional, Callable, TypeVar
from pathlib import Path
from contextlib import contextmanager
import logging

logger = logging.getLogger(__name__)

T = TypeVar('T')


class DatabaseConnection:
    """数据库连接上下文管理器"""

    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection: Optional[sqlite3.Connection] = None

    def __enter__(self) -> sqlite3.Connection:
        """进入上下文时建立连接"""
        self.connection = sqlite3.connect(self.db_path)
        self.connection.row_factory = sqlite3.Row
        logger.debug(f"Database connection established: {self.db_path}")
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        """退出上下文时关闭连接"""
        if self.connection:
            if exc_type is None:
                self.connection.commit()
                logger.debug("Transaction committed successfully")
            else:
                self.connection.rollback()
                logger.warning(f"Transaction rolled back due to: {exc_val}")

            self.connection.close()
            logger.debug("Database connection closed")

        return False


@contextmanager
def timer(name: str = "Operation"):
    """计时上下文管理器"""
    start_time = time.time()
    logger.info(f"{name} started")

    try:
        yield
    finally:
        elapsed = time.time() - start_time
        logger.info(f"{name} completed in {elapsed:.4f} seconds")


@contextmanager
def retry_on_failure(max_retries: int = 3, delay: float = 1.0, backoff: float = 2.0):
    """重试上下文管理器"""
    attempt = 0
    current_delay = delay

    while attempt < max_retries:
        try:
            yield attempt
            return
        except Exception as e:
            attempt += 1
            if attempt >= max_retries:
                logger.error(f"All {max_retries} attempts failed: {e}")
                raise

            logger.warning(f"Attempt {attempt} failed: {e}, retrying in {current_delay}s...")
            time.sleep(current_delay)
            current_delay *= backoff


class ResourceManager(Generic[T]):
    """通用资源管理器模板"""

    def __init__(
        self,
        acquire: Callable[[], T],
        release: Callable[[T], None]
    ):
        self._acquire = acquire
        self._release = release
        self._resource: Optional[T] = None

    def __enter__(self) -> T:
        self._resource = self._acquire()
        return self._resource

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        if self._resource is not None:
            self._release(self._resource)
        return False


# 使用示例
if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)

    # 使用数据库连接
    with DatabaseConnection(":memory:") as conn:
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
        cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
        print(f"Inserted user with id: {cursor.lastrowid}")

    # 使用计时器
    with timer("Data processing"):
        time.sleep(0.1)
        print("Processing complete")

    # 使用资源管理器
    manager = ResourceManager(
        acquire=lambda: {"data": [1, 2, 3]},
        release=lambda r: print(f"Released: {r}")
    )
    with manager as resource:
        print(f"Resource: {resource}")

# 异常处理装饰器库

import time
import functools
import logging
from typing import Callable, Any, Optional, Type, Tuple
import random
import asyncio

logger = logging.getLogger(__name__)


def retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,),
    on_retry: Optional[Callable[[Exception, int], None]] = None
):
    """
    重试装饰器 - 支持指数退避

    Args:
        max_attempts: 最大重试次数
        delay: 初始延迟（秒）
        backoff: 退避系数
        exceptions: 需要重试的异常类型元组
        on_retry: 每次重试时的回调函数
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            current_delay = delay
            last_exception = None

            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e

                    if attempt == max_attempts:
                        logger.error(
                            f"Function {func.__name__} failed after {max_attempts} attempts: {e}"
                        )
                        raise

                    if on_retry:
                        on_retry(e, attempt)

                    jitter = random.uniform(0.5, 1.5) if attempt > 1 else 1
                    sleep_time = current_delay * jitter

                    logger.warning(
                        f"Attempt {attempt}/{max_attempts} failed for {func.__name__}: {e}. "
                        f"Retrying in {sleep_time:.2f}s..."
                    )

                    time.sleep(sleep_time)
                    current_delay *= backoff

            raise last_exception

        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            current_delay = delay
            last_exception = None

            for attempt in range(1, max_attempts + 1):
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e

                    if attempt == max_attempts:
                        logger.error(
                            f"Async function {func.__name__} failed after {max_attempts} attempts: {e}"
                        )
                        raise

                    if on_retry:
                        on_retry(e, attempt)

                    jitter = random.uniform(0.5, 1.5) if attempt > 1 else 1
                    sleep_time = current_delay * jitter

                    logger.warning(
                        f"Attempt {attempt}/{max_attempts} failed for {func.__name__}: {e}. "
                        f"Retrying in {sleep_time:.2f}s..."
                    )

                    await asyncio.sleep(sleep_time)
                    current_delay *= backoff

            raise last_exception

        if asyncio.iscoroutinefunction(func):
            return async_wrapper
        return wrapper

    return decorator


def exception_handler(
    exceptions: Tuple[Type[Exception], ...] = (Exception,),
    default_value: Any = None,
    log_level: str = "error",
    reraise: bool = False
):
    """
    异常处理装饰器 - 统一处理特定异常
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except exceptions as e:
                log_func = getattr(logger, log_level)
                log_func(f"Exception in {func.__name__}: {e}", exc_info=True)

                if reraise:
                    raise

                return default_value() if callable(default_value) else default_value

        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            try:
                return await func(*args, **kwargs)
            except exceptions as e:
                log_func = getattr(logger, log_level)
                log_func(f"Exception in {func.__name__}: {e}", exc_info=True)

                if reraise:
                    raise

                return default_value() if callable(default_value) else default_value

        if asyncio.iscoroutinefunction(func):
            return async_wrapper
        return wrapper

    return decorator


def validate_input(**validators):
    """
    输入验证装饰器
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            import inspect
            sig = inspect.signature(func)
            bound = sig.bind(*args, **kwargs)
            bound.apply_defaults()

            for param_name, validator in validators.items():
                if param_name in bound.arguments:
                    value = bound.arguments[param_name]
                    if not validator(value):
                        raise ValueError(
                            f"Validation failed for parameter '{param_name}': {value}"
                        )

            return func(*args, **kwargs)

        return wrapper

    return decorator


def circuit_breaker(
    failure_threshold: int = 5,
    recovery_timeout: float = 60.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,)
):
    """
    断路器装饰器 - 防止级联故障
    """
    class CircuitBreakerState:
        def __init__(self):
            self.failure_count = 0
            self.last_failure_time: Optional[float] = None
            self.state: str = "closed"

    state = CircuitBreakerState()

    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            current_time = time.time()

            if state.state == "open":
                if state.last_failure_time and \
                   current_time - state.last_failure_time > recovery_timeout:
                    logger.info(f"Circuit breaker for {func.__name__} entering half-open state")
                    state.state = "half_open"
                else:
                    raise RuntimeError(f"Circuit breaker is open for {func.__name__}")

            try:
                result = func(*args, **kwargs)

                if state.state == "half_open":
                    logger.info(f"Circuit breaker for {func.__name__} closed")
                    state.state = "closed"
                    state.failure_count = 0

                return result

            except exceptions as e:
                state.failure_count += 1
                state.last_failure_time = current_time

                if state.failure_count >= failure_threshold:
                    state.state = "open"
                    logger.warning(
                        f"Circuit breaker opened for {func.__name__} after {failure_threshold} failures"
                    )

                raise

        return wrapper

    return decorator


# 使用示例
if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)

    @retry(max_attempts=3, delay=0.5, exceptions=(ConnectionError, TimeoutError))
    def call_api(url: str) -> dict:
        if random.random() < 0.7:
            raise ConnectionError("Network error")
        return {"status": "success", "data": [1, 2, 3]}

    result = call_api("https://api.example.com")
    print(f"API result: {result}")

    @exception_handler(ValueError, default_value={"error": "handled"})
    def parse_number(s: str) -> int:
        return int(s)

    print(f"Parse result: {parse_number('not a number')}")

    @validate_input(
        age=lambda x: isinstance(x, int) and 0 <= x <= 150,
        name=lambda x: isinstance(x, str) and len(x) >= 2
    )
    def create_user(name: str, age: int) -> dict:
        return {"name": name, "age": age}

    print(f"User created: {create_user('Alice', 30)}")

# 重试机制完整实现

import time
import random
import asyncio
import logging
from typing import Callable, Type, Tuple, Optional, Any, TypeVar
from dataclasses import dataclass
from enum import Enum

logger = logging.getLogger(__name__)

T = TypeVar('T')


class RetryState(Enum):
    """重试状态"""
    IDLE = "idle"
    RETRYING = "retrying"
    SUCCESS = "success"
    FAILED = "failed"
    EXHAUSTED = "exhausted"


@dataclass
class RetryResult:
    """重试结果"""
    state: RetryState
    attempts: int
    result: Any = None
    exception: Optional[Exception] = None
    total_time: float = 0.0


class ExponentialBackoff:
    """指数退避计算器"""

    def __init__(
        self,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        multiplier: float = 2.0,
        jitter: bool = True
    ):
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.multiplier = multiplier
        self.jitter = jitter

    def calculate(self, attempt: int) -> float:
        delay = self.base_delay * (self.multiplier ** (attempt - 1))
        delay = min(delay, self.max_delay)

        if self.jitter:
            jitter_factor = random.uniform(0.5, 1.5)
            delay *= jitter_factor

        return delay


class RetryPolicy:
    """可配置的重试策略"""

    def __init__(
        self,
        max_attempts: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        backoff_multiplier: float = 2.0,
        retry_on: Tuple[Type[Exception], ...] = (Exception,),
        on_retry: Optional[Callable[[Exception, int], None]] = None
    ):
        self.max_attempts = max_attempts
        self.backoff = ExponentialBackoff(
            base_delay=base_delay,
            max_delay=max_delay,
            multiplier=backoff_multiplier
        )
        self.retry_on = retry_on
        self.on_retry = on_retry

    def execute(self, func: Callable, *args, **kwargs) -> RetryResult:
        """执行带重试的函数"""
        start_time = time.time()

        for attempt in range(1, self.max_attempts + 1):
            try:
                result = func(*args, **kwargs)

                return RetryResult(
                    state=RetryState.SUCCESS,
                    attempts=attempt,
                    result=result,
                    total_time=time.time() - start_time
                )

            except Exception as e:
                if not isinstance(e, self.retry_on) or attempt >= self.max_attempts:
                    return RetryResult(
                        state=RetryState.FAILED,
                        attempts=attempt,
                        exception=e,
                        total_time=time.time() - start_time
                    )

                delay = self.backoff.calculate(attempt)

                logger.warning(
                    f"Attempt {attempt}/{self.max_attempts} failed: {e}. "
                    f"Retrying in {delay:.2f}s..."
                )

                if self.on_retry:
                    self.on_retry(e, attempt)

                time.sleep(delay)

        return RetryResult(
            state=RetryState.EXHAUSTED,
            attempts=self.max_attempts,
            total_time=time.time() - start_time
        )


# 使用示例
if __name__ == "__main__":
    def unstable_operation() -> str:
        if random.random() < 0.7:
            raise ConnectionError("Connection failed")
        return "Success!"

    policy = RetryPolicy(
        max_attempts=5,
        base_delay=0.5,
        backoff_multiplier=2.0,
        retry_on=(ConnectionError, TimeoutError)
    )

    result = policy.execute(unstable_operation)
    print(f"Result: {result}")

# logging 模块完整配置指南

import logging
import logging.handlers
import sys
from pathlib import Path
from typing import Optional
from datetime import datetime
import json


class JsonFormatter(logging.Formatter):
    """JSON 格式日志格式化器"""

    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno
        }

        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        if hasattr(record, "extra_data"):
            log_data["data"] = record.extra_data

        return json.dumps(log_data)


class ColoredFormatter(logging.Formatter):
    """带颜色的日志格式化器（终端输出）"""

    COLORS = {
        "DEBUG": "\033[36m",
        "INFO": "\033[32m",
        "WARNING": "\033[33m",
        "ERROR": "\033[31m",
        "CRITICAL": "\033[35m",
    }
    RESET = "\033[0m"

    def format(self, record: logging.LogRecord) -> str:
        color = self.COLORS.get(record.levelname, self.RESET)
        record.levelname = f"{color}{record.levelname}{self.RESET}"
        return super().format(record)


def setup_logging(
    level: str = "INFO",
    log_file: Optional[Path] = None,
    json_format: bool = False,
    max_bytes: int = 10 * 1024 * 1024,
    backup_count: int = 5
) -> logging.Logger:
    """配置日志系统"""
    logger = logging.getLogger()
    logger.setLevel(getattr(logging, level.upper()))
    logger.handlers.clear()

    # 控制台处理器
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.DEBUG)

    if json_format:
        console_formatter = JsonFormatter()
    else:
        console_formatter = ColoredFormatter(
            fmt="%(asctime)s | %(levelname)-18s | %(name)s:%(funcName)s:%(lineno)d | %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S"
        )

    console_handler.setFormatter(console_formatter)
    logger.addHandler(console_handler)

    # 文件处理器
    if log_file:
        log_file.parent.mkdir(parents=True, exist_ok=True)

        file_handler = logging.handlers.RotatingFileHandler(
            log_file,
            maxBytes=max_bytes,
            backupCount=backup_count,
            encoding="utf-8"
        )
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(JsonFormatter())
        logger.addHandler(file_handler)

    return logger


def get_logger(name: str, level: Optional[str] = None) -> logging.Logger:
    """获取指定名称的日志记录器"""
    logger = logging.getLogger(name)

    if level:
        logger.setLevel(getattr(logging, level.upper()))

    return logger


# 使用示例
if __name__ == "__main__":
    setup_logging(level="DEBUG", json_format=False)

    logger = get_logger(__name__)
    logger.debug("Debug message")
    logger.info("Info message")
    logger.warning("Warning message")
    logger.error("Error message")
    logger.critical("Critical message")

# 结构化日志系统

import logging
import json
from typing import Any, Dict, Optional
from datetime import datetime
from dataclasses import dataclass, field
from enum import Enum
import traceback


class LogLevel(Enum):
    DEBUG = "debug"
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"


@dataclass
class LogContext:
    """日志上下文"""
    request_id: Optional[str] = None
    user_id: Optional[str] = None
    session_id: Optional[str] = None
    correlation_id: Optional[str] = None
    extra: Dict[str, Any] = field(default_factory=dict)


class StructuredLogger:
    """结构化日志记录器"""

    def __init__(self, name: str, context: Optional[LogContext] = None):
        self.logger = logging.getLogger(name)
        self.context = context or LogContext()

    def _build_message(
        self,
        message: str,
        level: LogLevel,
        extra_data: Optional[Dict[str, Any]] = None,
        error: Optional[Exception] = None
    ) -> Dict[str, Any]:
        log_entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": level.value,
            "logger": self.logger.name,
            "message": message,
            "context": {}
        }

        if self.context.request_id:
            log_entry["context"]["request_id"] = self.context.request_id
        if self.context.user_id:
            log_entry["context"]["user_id"] = self.context.user_id

        if extra_data:
            log_entry["data"] = extra_data

        if error:
            log_entry["error"] = {
                "type": type(error).__name__,
                "message": str(error),
                "traceback": traceback.format_exc()
            }

        return log_entry

    def log(
        self,
        message: str,
        level: LogLevel,
        extra_data: Optional[Dict[str, Any]] = None,
        error: Optional[Exception] = None
    ):
        log_entry = self._build_message(message, level, extra_data, error)
        log_method = getattr(self.logger, level.value)
        log_method(json.dumps(log_entry))

    def debug(self, message: str, **extra_data):
        self.log(message, LogLevel.DEBUG, extra_data or None)

    def info(self, message: str, **extra_data):
        self.log(message, LogLevel.INFO, extra_data or None)

    def warning(self, message: str, **extra_data):
        self.log(message, LogLevel.WARNING, extra_data or None)

    def error(self, message: str, error: Optional[Exception] = None, **extra_data):
        self.log(message, LogLevel.ERROR, extra_data or None, error)

    def log_request(self, method: str, path: str, status_code: int, duration: float):
        """记录 HTTP 请求"""
        self.info(
            f"{method} {path} - {status_code}",
            method=method,
            path=path,
            status_code=status_code,
            duration_ms=round(duration * 1000, 2)
        )


class AuditLogger:
    """审计日志记录器"""

    def __init__(self, logger_name: str = "audit"):
        self.logger = logging.getLogger(logger_name)
        self.logger.setLevel(logging.INFO)

    def log_access(
        self,
        user_id: str,
        resource: str,
        action: str,
        success: bool,
        ip_address: Optional[str] = None
    ):
        entry = {
            "event_type": "access",
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "user_id": user_id,
            "resource": resource,
            "action": action,
            "success": success,
            "ip_address": ip_address
        }

        level = "INFO" if success else "WARNING"
        getattr(self.logger, level)(json.dumps(entry))

    def log_security_event(
        self,
        event_type: str,
        severity: str,
        description: str,
        user_id: Optional[str] = None,
        ip_address: Optional[str] = None
    ):
        entry = {
            "event_type": "security",
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "security_event_type": event_type,
            "severity": severity,
            "description": description,
            "user_id": user_id,
            "ip_address": ip_address
        }

        level = "WARNING" if severity in ("high", "critical") else "INFO"
        getattr(self.logger, level)(json.dumps(entry))


# 使用示例
if __name__ == "__main__":
    import sys
    logging.basicConfig(level=logging.DEBUG, handlers=[logging.StreamHandler(sys.stdout)])

    logger = StructuredLogger("myapp")
    logger.info("User logged in", user_name="alice")
    logger.warning("Rate limit approaching", current=95, limit=100)
    logger.error("Database connection failed", error=ConnectionError("DB unavailable"))

    audit = AuditLogger()
    audit.log_access(user_id="user_123", resource="/api/users", action="read", success=True)

# 日志级别与过滤

import logging
from typing import Set, Optional
from datetime import datetime, timedelta
import re


class SensitiveDataFilter(logging.Filter):
    """敏感数据过滤过滤器"""

    SENSITIVE_PATTERNS = [
        (re.compile(r'(password|passwd|pwd)[=:]\s*\S+', re.IGNORECASE), '[REDACTED]'),
        (re.compile(r'(token|api_key|apikey|secret)[=:]\s*\S+', re.IGNORECASE), '[REDACTED]'),
        (re.compile(r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b'), '****-****-****-****'),
    ]

    def filter(self, record: logging.LogRecord) -> bool:
        if record.msg:
            message = str(record.msg)
            for pattern, replacement in self.SENSITIVE_PATTERNS:
                message = pattern.sub(replacement, message)
            record.msg = message

        return True


class RateLimitFilter(logging.Filter):
    """速率限制过滤器"""

    def __init__(self, max_per_second: int = 10):
        super().__init__()
        self.max_per_second = max_per_second
        self.count = 0
        self.window_start = datetime.now()
        self.suppressed_count = 0

    def filter(self, record: logging.LogRecord) -> bool:
        now = datetime.now()

        if (now - self.window_start) > timedelta(seconds=1):
            if self.suppressed_count > 0:
                original_msg = record.msg
                record.msg = f"[Rate limited {self.suppressed_count} messages] {original_msg}"
                self.suppressed_count = 0

            self.count = 0
            self.window_start = now

        if self.count >= self.max_per_second:
            self.suppressed_count += 1
            return False

        self.count += 1
        return True


class ModuleLevelFilter(logging.Filter):
    """按模块过滤日志"""

    def __init__(self, allowed_modules: Optional[Set[str]] = None, blocked_modules: Optional[Set[str]] = None):
        super().__init__()
        self.allowed_modules = allowed_modules
        self.blocked_modules = blocked_modules or set()

    def filter(self, record: logging.LogRecord) -> bool:
        module_name = record.name

        if module_name in self.blocked_modules:
            return False

        if self.allowed_modules and module_name not in self.allowed_modules:
            return False

        return True


class LevelManager:
    """动态日志级别管理器"""

    def __init__(self):
        self._levels: dict = {}
        self._default_level = logging.INFO

    def set_level(self, logger_name: str, level: str):
        logger = logging.getLogger(logger_name)
        logger.setLevel(getattr(logging, level.upper()))
        self._levels[logger_name] = level.upper()

    def get_level(self, logger_name: str) -> str:
        return self._levels.get(logger_name, self._default_level)

    def set_all(self, level: str):
        self._default_level = level.upper()
        logging.getLogger().setLevel(self._default_level)


# 使用示例
if __name__ == "__main__":
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_handler.addFilter(SensitiveDataFilter())
    console_handler.addFilter(RateLimitFilter(max_per_second=5))

    root_logger = logging.getLogger()
    root_logger.addHandler(console_handler)
    root_logger.setLevel(logging.DEBUG)

    logger = logging.getLogger("test")
    logger.info("User login: password=secret123")

    for i in range(20):
        logger.info(f"Message {i}")

# 性能测试工具详解

import time
import timeit
import cProfile
import pstats
import io
import functools
from typing import Callable, Any, Dict
from dataclasses import dataclass


@dataclass
class BenchmarkResult:
    """基准测试结果"""
    name: str
    iterations: int
    total_time: float
    avg_time: float
    min_time: float
    max_time: float
    std_dev: float


class Benchmark:
    """基准测试框架"""

    def __init__(self, name: str = "Benchmark", iterations: int = 1000, warmup: int = 10):
        self.name = name
        self.iterations = iterations
        self.warmup = warmup
        self.results: Dict[str, BenchmarkResult] = {}

    def run(self, func: Callable, *args, **kwargs) -> BenchmarkResult:
        name = func.__name__

        # 预热
        for _ in range(self.warmup):
            func(*args, **kwargs)

        # 运行测试
        times = []
        for _ in range(self.iterations):
            start = time.perf_counter()
            func(*args, **kwargs)
            end = time.perf_counter()
            times.append(end - start)

        avg_time = sum(times) / len(times)
        min_time = min(times)
        max_time = max(times)

        variance = sum((t - avg_time) ** 2 for t in times) / len(times)
        std_dev = variance ** 0.5

        result = BenchmarkResult(
            name=name,
            iterations=self.iterations,
            total_time=sum(times),
            avg_time=avg_time,
            min_time=min_time,
            max_time=max_time,
            std_dev=std_dev
        )

        self.results[name] = result
        return result

    def compare(self) -> str:
        lines = [f"\n=== {self.name} Results ==="]
        lines.append(f"{'Function':<20} {'Avg (μs)':<15} {'Min (μs)':<15} {'Max (μs)':<15}")
        lines.append("-" * 65)

        for result in self.results.values():
            lines.append(
                f"{result.name:<20} "
                f"{result.avg_time*1e6:<15.2f} "
                f"{result.min_time*1e6:<15.2f} "
                f"{result.max_time*1e6:<15.2f}"
            )

        return "\n".join(lines)


def compare_implementations():
    """比较不同实现的性能"""

    benchmark = Benchmark("List Operations", iterations=1000, warmup=100)

    def loop_method(data):
        result = []
        for item in data:
            result.append(item * 2)
        return result

    def comprehension_method(data):
        return [item * 2 for item in data]

    def map_method(data):
        return list(map(lambda x: x * 2, data))

    test_data = list(range(10000))

    benchmark.run(loop_method, test_data)
    benchmark.run(comprehension_method, test_data)
    benchmark.run(map_method, test_data)

    print(benchmark.compare())


def cprofile_example():
    """cProfile 使用示例"""

    def slow_function():
        total = 0
        for i in range(10000):
            total += i ** 2
        return total

    def another_function():
        data = [i * 0.5 for i in range(1000)]
        return sorted(data)

    profiler = cProfile.Profile()
    profiler.enable()

    for _ in range(100):
        slow_function()
        another_function()

    profiler.disable()

    s = io.StringIO()
    ps = pstats.Stats(profiler, stream=s).sort_stats('cumulative')
    ps.print_stats(10)

    print("=== Profile Results ===")
    print(s.getvalue())


# 使用示例
if __name__ == "__main__":
    print("=== Benchmark Comparison ===")
    compare_implementations()

    print("\n=== cProfile Example ===")
    cprofile_example()