萬盛學電腦網

您现在的位置: 萬盛學電腦網 >> 程式設計 >  >> 網路程式設計 >> 程式語言綜合

Python 函數執行時間監控與日誌裝飾器:pre/post 模式與 context manager 實作

裝飾器模式:pre / post 兩種時機


from time import time
from functools import wraps

def logged(when: str = 'post'):
    """
    when='pre'  : 呼叫前記錄參數
    when='post' : 呼叫後記錄參數、返回值、耗時
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if when == 'pre':
                print(f"[PRE] {func.__name__} args={args} kwargs={kwargs}")
                return func(*args, **kwargs)
            elif when == 'post':
                start = time()
                try:
                    result = func(*args, **kwargs)
                    return result
                finally:
                    elapsed = time() - start
                    print(f"[POST] {func.__name__} args={args} kwargs={kwargs} -> {result} ({elapsed:.4f}s)")
            else:
                raise ValueError("when must be 'pre' or 'post'")
        return wrapper
    return decorator


# 使用範例
@logged('post')
def hello(name: str) -> str:
    time.sleep(0.1)
    return f"Hello, {name}!"

hello("World")
# [POST] hello args=('World',) kwargs={} -> Hello, World! (0.1003s)

進階:支援同步/非同步、結構化日誌、可配置 Logger


import logging
import asyncio
from functools import wraps
from time import perf_counter
from typing import Callable, Any

logger = logging.getLogger(__name__)

def timed(logger: logging.Logger = None, level: int = logging.INFO,
          log_args: bool = True, log_result: bool = False,
          log_exceptions: bool = True):
    """
    通用計時裝飾器,支援同步/非同步函數。
    """
    log = logger or logging.getLogger(__name__)

    def decorator(func: Callable) -> Callable:
        if asyncio.iscoroutinefunction(func):
            @wraps(func)
            async def async_wrapper(*args, **kwargs) -> Any:
                start = perf_counter()
                try:
                    result = await func(*args, **kwargs)
                    return result
                except Exception as e:
                    if log_exceptions:
                        log.log(level, f"{func.__qualname__} raised {type(e).__name__}: {e}", exc_info=True)
                    raise
                finally:
                    elapsed = perf_counter() - start
                    msg = f"{func.__qualname__} completed in {elapsed:.4f}s"
                    if log_args:
                        msg += f" | args={args} kwargs={kwargs}"
                    if log_result:
                        msg += f" -> {result}"
                    log.log(level, msg)
            return async_wrapper
        else:
            @wraps(func)
            def sync_wrapper(*args, **kwargs) -> Any:
                start = perf_counter()
                try:
                    result = func(*args, **kwargs)
                    return result
                except Exception as e:
                    if log_exceptions:
                        log.log(level, f"{func.__qualname__} raised {type(e).__name__}: {e}", exc_info=True)
                    raise
                finally:
                    elapsed = perf_counter() - start
                    msg = f"{func.__qualname__} completed in {elapsed:.4f}s"
                    if log_args:
                        msg += f" | args={args} kwargs={kwargs}"
                    if log_result:
                        msg += f" -> {result}"
                    log.log(level, msg)
            return sync_wrapper
    return decorator


# 使用
@timed(log_args=True, log_result=True)
def compute_sum(n: int) -> int:
    return sum(range(n))

@timed(level=logging.DEBUG)
async def fetch_data(url: str) -> dict:
    await asyncio.sleep(0.2)
    return {"url": url, "status": 200}

Context Manager 版本:適用於任意程式碼區塊


from contextlib import contextmanager
from time import perf_counter
import logging

@contextmanager
def timed_block(name: str, logger: logging.Logger = None, level: int = logging.INFO):
    log = logger or logging.getLogger(__name__)
    start = perf_counter()
    log.log(level, f"[{name}] 開始")
    try:
        yield
    except Exception as e:
        log.log(level, f"[{name}] 失敗: {e}", exc_info=True)
        raise
    finally:
        elapsed = perf_counter() - start
        log.log(level, f"[{name}] 完成,耗時 {elapsed:.4f}s")

# 使用
with timed_block("資料庫遷移"):
    run_migration()

# 嵌套計時
with timed_block("API 請求"):
    with timed_block("資料庫查詢"):
        query_db()
    with timed_block("外部服務呼叫"):
        call_external_api()

生產環境整合建議

  1. 結構化日誌:使用 python-json-logger 輸出 JSON,便於 ELK/Datadog/Loki 收集分析。
  2. 分散式追蹤:整合 OpenTelemetry,傳遞 trace_id/span_id 關聯上下游。
  3. 指標收集:搭配 prometheus_client 記錄 Histogram/Summary,而非僅靠日誌。
  4. 採樣:高頻介面啟用隨機採樣 (如 1%) 避免日誌洪水。

標準庫替代方案

  • timeit:微基準測試
  • cProfile / profile:完整效能剖析
  • functools.lru_cache:避免重複計算 (非計時但相關)
copyright © 萬盛學電腦網 all rights reserved