Giao diện
Decorators — Nghệ thuật mở rộng hàm không chạm code gốc
Mỗi endpoint trong Flask đều bắt đầu bằng @app.route. Mỗi test trong pytest đều sử dụng @pytest.fixture. Mỗi property trong class đều khai báo @property. Nếu bạn đã viết Python được vài tháng, bạn đã dùng decorator hàng trăm lần — nhưng có thể chưa bao giờ tự viết một cái. Decorator không phải cú pháp ma thuật, nó là một pattern cực kỳ thực tế: nhận hàm, trả về hàm mới, giữ nguyên hàm gốc.
Trong production, decorator giải quyết hàng loạt bài toán cross-cutting: logging, authentication, caching, rate limiting, retry logic, input validation. Thay vì copy-paste cùng một đoạn try-except vào 50 endpoint, bạn viết một decorator duy nhất và gắn lên mọi hàm cần thiết. Đó là nguyên tắc DRY ở dạng thuần khiết nhất — tách biệt what (logic nghiệp vụ) khỏi how (cơ chế bao quanh).
Bài viết này sẽ đưa bạn từ người dùng decorator thụ động thành người viết decorator chủ động. Bạn sẽ hiểu closure pattern đằng sau mọi decorator, nắm vững functools.wraps, xây dựng decorator factory có tham số, và triển khai decorator trong production với retry logic, exponential backoff, cùng proper logging.
Bức tranh tư duy
Hãy hình dung bạn có một món quà — đó chính là hàm gốc của bạn. Món quà đã hoàn chỉnh, bạn không muốn thay đổi nó. Nhưng bạn muốn thêm lớp giấy gói bên ngoài: giấy hoa, nơ, thiệp chúc mừng. Mỗi lớp giấy gói là một decorator — nó không chạm vào món quà bên trong, chỉ thêm hành vi phía ngoài.
┌─────────────────────────────────────────┐
│ @logging ← Lớp giấy gói ngoài │
│ ┌─────────────────────────────────┐ │
│ │ @authentication ← Lớp thứ 2 │ │
│ │ ┌─────────────────────────┐ │ │
│ │ │ @cache ← Lớp thứ 3 │ │ │
│ │ │ ┌───────────────┐ │ │ │
│ │ │ │ Hàm gốc 🎁 │ │ │ │
│ │ │ └───────────────┘ │ │ │
│ │ └─────────────────────────┘ │ │
│ └─────────────────────────────────┘ │
└─────────────────────────────────────────┘Khi ai đó "mở quà" (gọi hàm), họ phải đi qua từng lớp giấy gói từ ngoài vào trong: logging ghi nhận request → authentication kiểm tra quyền → cache kiểm tra kết quả cũ → cuối cùng mới tới hàm gốc. Mỗi lớp có quyền quyết định: cho đi tiếp, chặn lại, hoặc thay đổi kết quả trả về.
Một điểm quan trọng: thứ tự gói quà quyết định thứ tự xử lý. Decorator gần hàm nhất được áp dụng trước (gói trong cùng), decorator xa nhất được gọi trước khi hàm thực thi (lớp ngoài cùng). Đây là nguồn gốc của nhiều bug khi stacking decorator — chúng ta sẽ phân tích kỹ ở phần sau.
Cốt lõi kỹ thuật
Function decorator cơ bản — closure pattern
Mọi decorator đều dựa trên một nguyên lý: hàm trong Python là first-class object. Bạn có thể truyền hàm làm tham số, gán hàm vào biến, và trả về hàm từ hàm khác. Decorator tận dụng closure để "nhớ" hàm gốc bên trong wrapper.
python
def log_calls(func):
"""Decorator ghi log mỗi lần hàm được gọi."""
def wrapper(*args, **kwargs):
print(f"[LOG] Gọi {func.__name__} với args={args}, kwargs={kwargs}")
result = func(*args, **kwargs)
print(f"[LOG] {func.__name__} trả về {result}")
return result
return wrapper
@log_calls
def add(a, b):
"""Cộng hai số."""
return a + b
# Tương đương với: add = log_calls(add)
print(add(3, 5))
# [LOG] Gọi add với args=(3, 5), kwargs={}
# [LOG] add trả về 8
# 8Cú pháp @log_calls chỉ là syntactic sugar. Python đọc @log_calls phía trên def add và tự động thực thi add = log_calls(add). Biến add giờ trỏ tới wrapper, không còn trỏ tới hàm gốc nữa.
functools.wraps — tại sao bắt buộc phải dùng
Sau khi áp dụng decorator ở trên, hãy kiểm tra metadata:
python
print(add.__name__) # "wrapper" ← SAI! Phải là "add"
print(add.__doc__) # None ← SAI! Phải là "Cộng hai số."
help(add) # Help on function wrapper... ← Thảm họaDecorator đã "nuốt" identity của hàm gốc. Trong production, điều này gây ra: Sphinx documentation sai, pytest hiển thị tên test sai, Flask routing debug không ra endpoint nào, logging vô nghĩa vì mọi hàm đều tên "wrapper".
functools.wraps giải quyết triệt để vấn đề này:
python
import functools
def log_calls(func):
"""Decorator ghi log mỗi lần hàm được gọi."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
print(f"[LOG] Gọi {func.__name__} với args={args}, kwargs={kwargs}")
result = func(*args, **kwargs)
print(f"[LOG] {func.__name__} trả về {result}")
return result
return wrapper
@log_calls
def add(a, b):
"""Cộng hai số."""
return a + b
print(add.__name__) # "add" ✓
print(add.__doc__) # "Cộng hai số." ✓
print(add.__wrapped__) # <function add at 0x...> — truy cập hàm gốc@functools.wraps(func) copy các attribute __name__, __doc__, __dict__, __module__, __qualname__ từ hàm gốc sang wrapper. Ngoài ra nó thêm __wrapped__ trỏ về hàm gốc — cho phép bạn bypass decorator khi cần (ví dụ trong unit test).
Decorator có tham số — decorator factory
Khi decorator cần nhận cấu hình, bạn cần thêm một lớp hàm bao ngoài nữa. Đây gọi là decorator factory — một hàm trả về decorator:
python
import functools
import time
def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
"""Decorator factory: retry hàm khi gặp exception."""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as exc:
last_exception = exc
if attempt < max_attempts:
print(
f"[RETRY] {func.__name__} lần {attempt}/{max_attempts}, "
f"chờ {delay}s: {exc}"
)
time.sleep(delay)
raise last_exception
return wrapper
return decorator
@retry(max_attempts=5, delay=2.0, exceptions=(ConnectionError, TimeoutError))
def fetch_user_data(user_id):
"""Gọi API lấy dữ liệu người dùng."""
import random
if random.random() < 0.7:
raise ConnectionError("Server không phản hồi")
return {"id": user_id, "name": "Nguyễn Văn A"}Luồng thực thi: retry(max_attempts=5, ...) trả về decorator → decorator(fetch_user_data) trả về wrapper → wrapper là hàm cuối cùng được gán cho fetch_user_data.
Ba tầng hàm lồng nhau là pattern chuẩn. Nếu bạn thấy khó đọc, hãy nhớ quy tắc: factory nhận config → decorator nhận func → wrapper nhận args.
Class decorator — stateful decorator với __call__
Khi decorator cần lưu trạng thái giữa các lần gọi (ví dụ: đếm số lần gọi, cache kết quả), class decorator là lựa chọn sạch hơn closure:
python
import functools
class CountCalls:
"""Decorator đếm số lần hàm được gọi."""
def __init__(self, func):
functools.update_wrapper(self, func)
self.func = func
self.call_count = 0
def __call__(self, *args, **kwargs):
self.call_count += 1
print(f"[COUNT] {self.func.__name__} đã được gọi {self.call_count} lần")
return self.func(*args, **kwargs)
def reset(self):
"""Reset bộ đếm về 0."""
self.call_count = 0
@CountCalls
def process_order(order_id):
"""Xử lý đơn hàng."""
return f"Đơn hàng {order_id} đã xử lý"
process_order("ORD-001") # [COUNT] process_order đã được gọi 1 lần
process_order("ORD-002") # [COUNT] process_order đã được gọi 2 lần
print(process_order.call_count) # 2 — truy cập state từ bên ngoài
process_order.reset() # Reset bộ đếmClass decorator cũng có thể dùng để decorate class (không chỉ function). Ví dụ thêm method hoặc attribute vào class:
python
def add_repr(cls):
"""Decorator thêm __repr__ tự động cho class."""
def __repr__(self):
attrs = ", ".join(f"{k}={v!r}" for k, v in self.__dict__.items())
return f"{cls.__name__}({attrs})"
cls.__repr__ = __repr__
return cls
@add_repr
class User:
def __init__(self, name, email):
self.name = name
self.email = email
print(User("Minh", "minh@example.com"))
# User(name='Minh', email='minh@example.com')Stacking decorators — thứ tự thực thi
Khi stack nhiều decorator, thứ tự áp dụng (từ dưới lên) ngược với thứ tự thực thi (từ trên xuống):
python
import functools
def bold(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
return f"<b>{func(*args, **kwargs)}</b>"
return wrapper
def italic(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
return f"<i>{func(*args, **kwargs)}</i>"
return wrapper
@bold # Bước 2: bold(italic_wrapper) → bold_wrapper
@italic # Bước 1: italic(greet) → italic_wrapper
def greet(name):
return f"Xin chào {name}"
print(greet("Python"))
# <b><i>Xin chào Python</i></b>
# ↑ bold ở ngoài, italic ở trongQuy tắc ghi nhớ: decorator gần hàm nhất được gói trong cùng (thực thi cuối), decorator xa nhất bọc ngoài cùng (thực thi đầu). Điều này tương đương với:
python
greet = bold(italic(greet))Trong production, thứ tự stacking rất quan trọng. Ví dụ: @login_required phải nằm trên @cache — bạn không muốn cache response của user chưa đăng nhập rồi trả về cho user đã đăng nhập.
Thực chiến
Retry decorator với exponential backoff cho FastAPI
Trong production, API call tới service bên ngoài thường xuyên thất bại tạm thời (network timeout, rate limit, server overload). Retry với exponential backoff là pattern tiêu chuẩn. Dưới đây là implementation production-grade:
python
import asyncio
import functools
import logging
import random
import time
from typing import Tuple, Type
logger = logging.getLogger(__name__)
def retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True,
retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,),
):
"""
Decorator factory: retry với exponential backoff.
Args:
max_retries: Số lần retry tối đa (không tính lần gọi đầu).
base_delay: Thời gian chờ cơ sở (giây).
max_delay: Thời gian chờ tối đa (giây).
exponential_base: Hệ số nhân cho mỗi lần retry.
jitter: Thêm random jitter để tránh thundering herd.
retryable_exceptions: Tuple các exception class được phép retry.
"""
def decorator(func):
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries + 1):
try:
return await func(*args, **kwargs)
except retryable_exceptions as exc:
last_exception = exc
if attempt == max_retries:
logger.error(
"[RETRY] %s thất bại sau %d lần thử: %s",
func.__name__, max_retries + 1, exc,
)
raise
delay = min(
base_delay * (exponential_base ** attempt),
max_delay,
)
if jitter:
delay = delay * (0.5 + random.random())
logger.warning(
"[RETRY] %s lần %d/%d, chờ %.2fs: %s",
func.__name__, attempt + 1, max_retries,
delay, exc,
)
await asyncio.sleep(delay)
raise last_exception
@functools.wraps(func)
def sync_wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except retryable_exceptions as exc:
last_exception = exc
if attempt == max_retries:
logger.error(
"[RETRY] %s thất bại sau %d lần thử: %s",
func.__name__, max_retries + 1, exc,
)
raise
delay = min(
base_delay * (exponential_base ** attempt),
max_delay,
)
if jitter:
delay = delay * (0.5 + random.random())
logger.warning(
"[RETRY] %s lần %d/%d, chờ %.2fs: %s",
func.__name__, attempt + 1, max_retries,
delay, exc,
)
time.sleep(delay)
raise last_exception
if asyncio.iscoroutinefunction(func):
return async_wrapper
return sync_wrapper
return decoratorSử dụng trong FastAPI:
python
import httpx
from fastapi import FastAPI, HTTPException
app = FastAPI()
@retry_with_backoff(
max_retries=3,
base_delay=0.5,
retryable_exceptions=(httpx.TimeoutException, httpx.ConnectError),
)
async def call_payment_gateway(order_id: str, amount: float) -> dict:
"""Gọi payment gateway bên ngoài."""
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.post(
"https://payment.example.com/charge",
json={"order_id": order_id, "amount": amount},
)
response.raise_for_status()
return response.json()
@app.post("/orders/{order_id}/pay")
async def process_payment(order_id: str, amount: float):
try:
result = await call_payment_gateway(order_id, amount)
return {"status": "success", "transaction": result}
except httpx.TimeoutException:
raise HTTPException(status_code=504, detail="Payment gateway timeout")
except httpx.ConnectError:
raise HTTPException(status_code=502, detail="Payment gateway unreachable")Lưu ý thiết kế: decorator tự phát hiện hàm async hay sync nhờ asyncio.iscoroutinefunction() và chọn wrapper phù hợp. Jitter ngẫu nhiên ngăn chặn thundering herd — khi nhiều request cùng retry đồng thời, jitter phân tán thời điểm retry để tránh tạo spike lên server đích.
Sai lầm điển hình
Sai lầm 1: Quên @functools.wraps — mất metadata
❌ SAI:
python
def timer(func):
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
print(f"{func.__name__}: {time.time() - start:.4f}s")
return result
return wrapper
@timer
def calculate_report(data):
"""Tạo báo cáo tài chính."""
...
# Hậu quả production:
print(calculate_report.__name__) # "wrapper" → logging vô nghĩa
print(calculate_report.__doc__) # None → Sphinx docs trống
# pytest hiển thị: PASSED test_wrapper → không biết test gì
# Flask debug: Endpoint "wrapper" → không trace được route✅ ĐÚNG:
python
import functools
def timer(func):
@functools.wraps(func) # ← Một dòng cứu cả project
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
print(f"{func.__name__}: {time.time() - start:.4f}s")
return result
return wrapperImpact: Trong project có 200+ decorator, thiếu @wraps khiến mọi hàm trong log đều tên "wrapper". Debug incident lúc nửa đêm trở thành ác mộng.
Sai lầm 2: Decorator không trả về wrapper — hàm trả về None
❌ SAI:
python
def validate_input(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if not args:
raise ValueError("Thiếu argument")
func(*args, **kwargs) # ← Quên return!
return wrapper
@validate_input
def get_user(user_id):
return {"id": user_id, "name": "Minh"}
result = get_user(42)
print(result) # None ← Bug âm thầm, không có exception!✅ ĐÚNG:
python
def validate_input(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if not args:
raise ValueError("Thiếu argument")
return func(*args, **kwargs) # ← return rõ ràng
return wrapperImpact: Bug này đặc biệt nguy hiểm vì không raise exception. Hàm chạy đúng logic bên trong nhưng kết quả bị "nuốt" — response trả về None, data bị mất, downstream service nhận giá trị sai.
Sai lầm 3: Nhầm thứ tự stacking decorator
❌ SAI:
python
@cache_response # Cache TRƯỚC khi check auth
@require_authentication # Auth check ở trong
def get_dashboard(user_id):
return fetch_dashboard_data(user_id)
# Kịch bản: User A (admin) gọi → cache lưu kết quả admin
# User B (guest) gọi → cache trả kết quả admin cho guest!
# → Lỗ hổng bảo mật nghiêm trọng✅ ĐÚNG:
python
@require_authentication # Auth check TRƯỚC
@cache_response # Cache SAU khi đã xác thực
def get_dashboard(user_id):
return fetch_dashboard_data(user_id)Impact: Sai thứ tự stacking có thể tạo lỗ hổng bảo mật (cache bypass authentication), hoặc logic sai (logging ghi nhận request sau khi bị reject, rate limit đếm request đã bị cache).
Sai lầm 4: Decorator có side effect khi import
❌ SAI:
python
import time
def register_endpoint(func):
# Side effect ngay khi module được import!
print(f"Đăng ký endpoint: {func.__name__} lúc {time.time()}")
API_REGISTRY[func.__name__] = func # Modify global state
return func
# Mỗi khi bất kỳ module nào import file này,
# tất cả decorator đều chạy side effect — kể cả trong test!✅ ĐÚNG:
python
import functools
import inspect
def register_endpoint(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
wrapper._is_endpoint = True # Metadata, không phải side effect
return wrapper
def discover_endpoints(module):
"""Registration riêng biệt, kiểm soát được thời điểm."""
for name, obj in inspect.getmembers(module):
if getattr(obj, "_is_endpoint", False):
API_REGISTRY[name] = objImpact: Side effect khi import gây test flaky, circular import, và hành vi không dự đoán được khi thứ tự import thay đổi. Đặc biệt nguy hiểm với auto-reload trong development server.
Under the Hood
Decorator ở tầng bytecode
Khi CPython biên dịch decorator syntax, nó tạo ra bytecode tương đương với phép gán. Hãy xem dis output:
python
import dis
def my_decorator(func):
def wrapper():
return func()
return wrapper
# Xem bytecode khi CPython xử lý @my_decorator trước def hello:
# LOAD_GLOBAL my_decorator ← Load decorator function
# LOAD_CONST <code object> ← Load code của hello
# MAKE_FUNCTION 0 ← Tạo function object cho hello
# CALL_FUNCTION 1 ← Gọi my_decorator(hello)
# STORE_NAME hello ← Gán kết quả vào tên "hello"Decorator không phải cơ chế đặc biệt ở tầng bytecode — nó chỉ là syntactic sugar cho một lời gọi hàm. @decorator trước def func biên dịch thành func = decorator(func) sau khi hàm được tạo.
Closure cells và biến tự do
Wrapper function trong decorator là một closure — nó "bắt" (capture) biến func từ scope bao ngoài:
python
import functools
def multiplier(factor):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs) * factor
return wrapper
return decorator
@multiplier(3)
def get_base_price(item_id):
return 100
# Kiểm tra closure internals:
print(get_base_price.__code__.co_freevars)
# ('factor', 'func') — các biến tự do được capture
print(get_base_price.__closure__)
# (<cell at 0x...: int object at 0x...>,
# <cell at 0x...: function object at 0x...>)
# Truy cập giá trị trong cell:
print(get_base_price.__closure__[0].cell_contents) # 3 (factor)
print(get_base_price.__closure__[1].cell_contents) # <function get_base_price>CPython lưu closure variables trong cell objects. Mỗi cell là một con trỏ gián tiếp đến giá trị thực — cho phép nhiều closure chia sẻ cùng một biến (quan trọng khi closure được tạo trong vòng lặp).
Chi phí hiệu năng của decorator
Mỗi lớp decorator thêm một frame call vào call stack. Benchmark điển hình (CPython 3.12, 1 triệu iterations):
| Cấu hình | Thời gian | Overhead |
|---|---|---|
| Raw function | 0.062s | baseline |
| 1 decorator layer | 0.098s | +58% |
| 3 decorator layers | 0.172s | +177% |
Overhead ~30-60ns mỗi lớp. Không đáng lo cho web endpoint (latency ms), nhưng đáng cân nhắc cho tight loop trong data processing.
functools.wraps hoạt động thế nào
@functools.wraps(func) thực chất gọi functools.update_wrapper(wrapper, func), copy các attribute sau:
python
# Danh sách attribute được copy (WRAPPER_ASSIGNMENTS):
# __module__ — module chứa hàm gốc
# __name__ — tên hàm
# __qualname__ — tên đầy đủ (class.method)
# __doc__ — docstring
# __dict__ — attribute dictionary
# __annotations__ — type annotations
# Attribute được update (WRAPPER_UPDATES):
# __dict__ — merge dict của hàm gốc vào wrapper
# Attribute bổ sung:
# __wrapped__ — reference đến hàm gốc (cho phép bypass decorator)Attribute __wrapped__ đặc biệt hữu ích trong testing — bạn có thể test hàm gốc mà không cần đi qua decorator:
python
# Trong test, bypass decorator để test pure logic:
original_func = my_decorated_function.__wrapped__
result = original_func(test_input)
assert result == expected_outputChecklist ghi nhớ
✅ Checklist triển khai
- [ ] Luôn dùng
@functools.wraps(func)trong mọi wrapper function - [ ] Decorator phải trả về wrapper (hoặc hàm gốc) — không bao giờ trả về
Nonengầm - [ ] Wrapper phải nhận
*args, **kwargsđể tương thích với mọi signature - [ ] Wrapper phải
return func(*args, **kwargs)trừ khi cố ý thay đổi kết quả - [ ] Decorator factory cần 3 tầng: factory(config) → decorator(func) → wrapper(args)
- [ ] Kiểm tra thứ tự stacking: decorator ngoài cùng thực thi trước nhất
- [ ] Tránh side effect trong decorator body — chỉ có side effect trong wrapper
- [ ] Class decorator dùng
functools.update_wrapper(self, func)trong__init__ - [ ] Test decorator riêng biệt: test wrapper behavior VÀ test hàm gốc qua
__wrapped__ - [ ] Decorator cho async function phải trả về async wrapper (dùng
asyncio.iscoroutinefunction) - [ ] Ghi log trong decorator nên dùng
loggingmodule, không dùngprint - [ ] Decorator nên có docstring giải thích mục đích và tham số
- [ ] Profile performance nếu decorator được gọi trong tight loop (>100K calls/s)
- [ ] Đừng lạm dụng decorator — nếu logic phức tạp, dùng explicit function call rõ ràng hơn
Bài tập luyện tập
Bài 1 — @timer decorator (Foundation)
Viết decorator @timer đo thời gian thực thi của hàm, in kết quả ra console với format: [TIMER] function_name: 0.1234s
Yêu cầu:
- Dùng
functools.wraps - Dùng
time.perf_counter()(chính xác hơntime.time()) - Trả về kết quả của hàm gốc
🧠 Quiz — Kiểm tra nhanh
Câu hỏi: Nếu decorator không có @functools.wraps(func), attribute nào của hàm gốc sẽ bị mất?
- A. Chỉ
__name__ - B. Chỉ
__doc__ - C.
__name__,__doc__,__module__,__qualname__,__annotations__ - D. Không mất attribute nào, Python tự bảo toàn
Đáp án
C — Khi không dùng @functools.wraps, wrapper thay thế hoàn toàn hàm gốc. Tất cả metadata (__name__, __doc__, __module__, __qualname__, __annotations__) đều trỏ về wrapper, không còn thông tin của hàm gốc. functools.wraps copy chính xác các attribute này từ hàm gốc sang wrapper.
:::
Lời giải Bài 1
python
import functools
import time
def timer(func):
"""Decorator đo thời gian thực thi."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"[TIMER] {func.__name__}: {elapsed:.4f}s")
return result
return wrapper
# Kiểm thử:
@timer
def slow_function():
"""Hàm chạy chậm để test."""
time.sleep(0.5)
return "done"
result = slow_function()
assert result == "done"
assert slow_function.__name__ == "slow_function"
assert slow_function.__doc__ == "Hàm chạy chậm để test."
print("✓ Tất cả assertions passed")Bài 2 — @rate_limit decorator (Intermediate)
Viết decorator factory @rate_limit(max_calls, period) giới hạn số lần gọi hàm trong một khoảng thời gian.
Yêu cầu:
max_calls: số lần gọi tối đaperiod: khoảng thời gian (giây)- Raise
RateLimitExceededkhi vượt giới hạn - Thread-safe (dùng
threading.Lock) - Tự động reset sau khi hết period
🧠 Quiz — Kiểm tra nhanh
Câu hỏi: Trong decorator factory pattern @rate_limit(max_calls=10, period=60), có bao nhiêu tầng hàm lồng nhau?
- A. 1 tầng: wrapper
- B. 2 tầng: decorator → wrapper
- C. 3 tầng: factory → decorator → wrapper
- D. 4 tầng: outer → factory → decorator → wrapper
Đáp án
C — Decorator factory luôn có 3 tầng:
rate_limit(max_calls, period)— factory, nhận cấu hìnhdecorator(func)— decorator thực sự, nhận hàm cần wrapwrapper(*args, **kwargs)— hàm thay thế, nhận argument của hàm gốc
Khi viết @rate_limit(max_calls=10), Python gọi rate_limit(10) → nhận decorator → gọi decorator(func) → nhận wrapper → gán wrapper cho tên hàm.
:::
Lời giải Bài 2
python
import functools
import threading
import time
from collections import deque
class RateLimitExceeded(Exception):
"""Exception khi vượt giới hạn tần suất gọi."""
pass
def rate_limit(max_calls: int, period: float):
"""
Decorator factory giới hạn tần suất gọi hàm.
Args:
max_calls: Số lần gọi tối đa trong khoảng period.
period: Khoảng thời gian (giây) để tính rate limit.
"""
def decorator(func):
call_times = deque()
lock = threading.Lock()
@functools.wraps(func)
def wrapper(*args, **kwargs):
with lock:
now = time.monotonic()
# Loại bỏ các timestamp cũ hơn period
while call_times and call_times[0] <= now - period:
call_times.popleft()
if len(call_times) >= max_calls:
oldest = call_times[0]
retry_after = period - (now - oldest)
raise RateLimitExceeded(
f"{func.__name__} bị giới hạn: "
f"{max_calls} lần/{period}s. "
f"Thử lại sau {retry_after:.1f}s."
)
call_times.append(now)
return func(*args, **kwargs)
wrapper.reset = lambda: call_times.clear()
return wrapper
return decorator
# Kiểm thử:
@rate_limit(max_calls=3, period=1.0)
def api_call(endpoint):
"""Gọi API endpoint."""
return f"Response từ {endpoint}"
# 3 lần đầu OK
for i in range(3):
print(api_call(f"/users/{i}"))
# Lần thứ 4 bị chặn
try:
api_call("/users/3")
except RateLimitExceeded as exc:
print(f"✓ Rate limited: {exc}")
# Chờ hết period rồi gọi lại OK
time.sleep(1.1)
print(api_call("/users/4"))
print("✓ Tất cả assertions passed")