Skip to content

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
# 8

Cú 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ọa

Decorator đã "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ề decoratordecorator(fetch_user_data) trả về wrapperwrapper 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ộ đếm

Class 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 ở trong

Quy 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 decorator

Sử 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 wrapper

Impact: 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 wrapper

Impact: 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] = obj

Impact: 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ìnhThời gianOverhead
Raw function0.062sbaseline
1 decorator layer0.098s+58%
3 decorator layers0.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_output

Checklist 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ề None ngầ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 logging module, không dùng print
  • [ ] 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ơn time.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 đa
  • period: khoảng thời gian (giây)
  • Raise RateLimitExceeded khi 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:

  1. rate_limit(max_calls, period) — factory, nhận cấu hình
  2. decorator(func) — decorator thực sự, nhận hàm cần wrap
  3. wrapper(*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")

Liên kết học tiếp