裝飾器模式: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()
生產環境整合建議
- 結構化日誌:使用
python-json-logger 輸出 JSON,便於 ELK/Datadog/Loki 收集分析。
- 分散式追蹤:整合 OpenTelemetry,傳遞
trace_id/span_id 關聯上下游。
- 指標收集:搭配
prometheus_client 記錄 Histogram/Summary,而非僅靠日誌。
- 採樣:高頻介面啟用隨機採樣 (如 1%) 避免日誌洪水。
標準庫替代方案
timeit:微基準測試
cProfile / profile:完整效能剖析
functools.lru_cache:避免重複計算 (非計時但相關)