Decorator và Context Manager — hai metaprogramming tool Pythonic

1. Decorator — mental model

Decorator trong Python không phải "Decorator pattern" của Gang of Four. Tên trùng nhau nhưng khác concept. Decorator của Python là một callable (thường là function) nhận một callable và trả về một callable. Mục đích: bọc thêm hành vi (logging, timing, retry, caching, validation) mà không phải sửa code function gốc.

Syntax @decorator ở trên dòng def chỉ là syntax sugar:

# Đây:
@decorator
def f():
    pass

# Tương đương:
def f():
    pass
f = decorator(f)  # gán đè biến f bằng kết quả decorator(f)

Nói cách khác, sau khi def f evaluate xong, Python lập tức gọi decorator(f) và gán kết quả vào tên f. Kết quả thường là một function khác (gọi là wrapper) bao quanh function gốc.

Quan trọng — decorator chạy tại thời điểm def được evaluate, không phải khi gọi f():

def noisy(fn):
    print(f"decorator đang wrap {fn.__name__}")
    return fn

@noisy
def hello():
    print("hello")

# Đã in ra "decorator đang wrap hello" — NGAY KHI def evaluate
# Chưa hề gọi hello()

hello()  # Bây giờ in "hello"

2. Viết decorator đơn giản

Decorator điển hình có 2 lớp: outer (nhận fn) và inner (wrapper thực sự được gọi). Inner phải chấp nhận mọi argument bằng *args, **kwargs để wrap được mọi signature.

def trace(fn):
    def wrapper(*args, **kwargs):
        print(f"→ calling {fn.__name__}({args}, {kwargs})")
        result = fn(*args, **kwargs)
        print(f"← {fn.__name__} trả về {result!r}")
        return result
    return wrapper

@trace
def add(a, b):
    return a + b

add(2, 3)
# → calling add((2, 3), {})
# ← add trả về 5

Phân tích từng dòng:

• def trace(fn) — outer function, nhận function được wrap.
• def wrapper(*args, **kwargs) — inner function, sẽ thay add. Signature linh hoạt để wrap mọi function.
• fn(*args, **kwargs) — gọi function gốc với đúng args caller truyền vào.
• return wrapper — outer trả về wrapper. @trace khiến add = trace(add) = wrapper.

Vì wrapper capture biến fn qua closure (xem Chương 3), function gốc không bị mất.

3. functools.wraps — bảo toàn metadata

Có một vấn đề tinh tế với decorator. Sau khi wrap, add không còn trỏ vào function gốc nữa, mà trỏ vào wrapper. Hệ quả: metadata như __name__, __doc__, __module__, __annotations__ của add bị thay bằng của wrapper:

def trace(fn):
    def wrapper(*args, **kwargs):
        return fn(*args, **kwargs)
    return wrapper

@trace
def add(a, b):
    """Cộng 2 số."""
    return a + b

print(add.__name__)  # 'wrapper' 🔥 (đáng lẽ 'add')
print(add.__doc__)   # None       🔥 (mất docstring)
help(add)            # Help on function wrapper... 🔥

Điều này phá vỡ help(), debugger, profiler, tài liệu tự sinh (Sphinx). Fix bằng @functools.wraps(fn) trên wrapper — copy metadata từ fn sang wrapper:

import functools

def trace(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        return fn(*args, **kwargs)
    return wrapper

@trace
def add(a, b):
    """Cộng 2 số."""
    return a + b

print(add.__name__)      # 'add'         ✓
print(add.__doc__)       # 'Cộng 2 số.'  ✓
print(add.__wrapped__)   # <function add> ← reference function gốc

functools.wraps copy: __module__, __name__, __qualname__, __annotations__, __doc__, và set __wrapped__ trỏ về fn gốc. Cái __wrapped__ hữu ích khi cần "unwrap" decorator để introspect function gốc (vd: inspect.signature dùng nó).

4. Decorator có argument — decorator factory

Đôi khi bạn cần decorator nhận tham số: @repeat(3), @retry(times=3), @route('/users'). Cần thêm 1 lớp ngoài: decorator factory nhận argument, trả về decorator thật.

Phân tích chuyển hoá:

@repeat(3)
def greet():
    print("hi")

# Tương đương:
def greet():
    print("hi")
greet = repeat(3)(greet)
#        ↑ gọi factory   ↑ gọi decorator trả về

Vậy repeat(3) phải trả về một decorator (function nhận function trả function). Cấu trúc 3 tầng:

import functools

def repeat(n):                       # Tầng 1: factory, nhận args
    def decorator(fn):               # Tầng 2: decorator thật, nhận fn
        @functools.wraps(fn)
        def wrapper(*args, **kwargs): # Tầng 3: wrapper, replace fn
            result = None
            for _ in range(n):
                result = fn(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(3)
def greet(name):
    print(f"hi {name}")

greet("An")
# hi An
# hi An
# hi An

@repeat          # f = repeat(f) — repeat nhận function!
@repeat()        # f = repeat()(f) — repeat nhận () rồi mới wrap
@repeat(3)       # f = repeat(3)(f)

5. Stack decorator — thứ tự apply

Bạn có thể chồng nhiều decorator lên 1 function. Thứ tự apply là bottom-up (từ dưới lên): decorator gần def nhất apply trước.

@a
@b
@c
def f():
    pass

# Tương đương:
def f():
    pass
f = a(b(c(f)))
#     ↑ c apply trước (gần def nhất)
#   ↑ rồi b
# ↑ rồi a (ngoài cùng)

Khi gọi f(x), control flow là top-down: a chạy trước, gọi b, rồi b gọi c, rồi c gọi function gốc. Trả về thì reverse.

def make_dec(label):
    def dec(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kw):
            print(f"{label} trước")
            result = fn(*args, **kw)
            print(f"{label} sau")
            return result
        return wrapper
    return dec

@make_dec("A")
@make_dec("B")
@make_dec("C")
def f():
    print("  f")

f()
# A trước
# B trước
# C trước
#   f
# C sau
# B sau
# A sau

Order matters! Vd: @cache ngoài @trace vs @trace ngoài @cache — kết quả khác:

• @cache trên cùng → cache hit không in trace (vì cache trả ngay, không gọi xuống).
• @trace trên cùng → mỗi call đều log, kể cả cache hit.

6. functools.cache / lru_cache — memoization

Memoization là tối ưu kinh điển: lưu kết quả function theo argument, lần sau cùng args trả ngay từ cache. Python có sẵn 2 decorator:

• @functools.cache — Python 3.9+. Cache không giới hạn.
• @functools.lru_cache(maxsize=128) — Có giới hạn (LRU eviction). Default 128.

import functools
import time

# Không cache: exponential O(2^n)
def fib_slow(n):
    if n < 2:
        return n
    return fib_slow(n - 1) + fib_slow(n - 2)

t = time.perf_counter()
fib_slow(35)               # ~2-3 giây
print(time.perf_counter() - t)

# Có cache: linear O(n) (sau khi memoize)
@functools.cache
def fib_fast(n):
    if n < 2:
        return n
    return fib_fast(n - 1) + fib_fast(n - 2)

t = time.perf_counter()
fib_fast(500)              # vài microseconds
print(time.perf_counter() - t)
print(fib_fast.cache_info())  # CacheInfo(hits=..., misses=..., ...)

Hai biến thể có signature cache giống nhau, chỉ khác policy eviction:

7. Class-based decorator

Decorator không nhất thiết là function. Bất cứ object nào callable đều dùng được. Một class với __call__ là decorator hợp lệ. Hữu ích khi cần giữ state giữa các lần gọi.

import functools

class CountCalls:
    def __init__(self, fn):
        functools.update_wrapper(self, fn)  # thay cho @wraps
        self.fn = fn
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"Call #{self.count} to {self.fn.__name__}")
        return self.fn(*args, **kwargs)

@CountCalls
def greet(name):
    print(f"Hi {name}")

greet("An")   # Call #1 to greet
greet("Bình") # Call #2 to greet
print(greet.count)  # 2 — truy cập state qua attribute

Khi @CountCalls apply, Python gọi CountCalls(greet) → tạo instance, gán vào greet. Mỗi greet(...) thực ra là instance(...) → trigger __call__.

Class-based decorator có argument (decorator factory) khác chút:

class Retry:
    def __init__(self, times):
        self.times = times

    def __call__(self, fn):
        @functools.wraps(fn)
        def wrapper(*args, **kw):
            for i in range(self.times):
                try:
                    return fn(*args, **kw)
                except Exception:
                    if i == self.times - 1: raise
        return wrapper

@Retry(times=3)
def flaky():
    ...

8. Decorator phổ biến trong stdlib và lib

Decorator có mặt khắp Python ecosystem. Một số tiêu biểu bạn sẽ gặp hằng ngày:

Khi đọc framework code thấy @something, hỏi 3 câu: (1) nó wrap function thành cái gì khác? (2) chạy lúc nào — at-import hay at-call? (3) thêm state/side-effect gì? Nắm 3 câu là hiểu được decorator bất kỳ.

9. Context manager với with statement

Context manager giải quyết bài toán: "luôn cleanup resource khi xong, kể cả khi exception". Resource có thể là file, socket, lock, database connection, transaction, temp directory.

Cách viết "thủ công" với try/finally:

f = open("data.txt")
try:
    content = f.read()
    process(content)        # nếu cái này throw...
finally:
    f.close()               # file vẫn được close. Tốt nhưng verbose.

Cách Pythonic với with:

with open("data.txt") as f:
    content = f.read()
    process(content)
# f.close() được gọi tự động khi ra khỏi with, kể cả khi exception

Cú pháp:

• with EXPR as VAR: — EXPR là biểu thức trả về context manager. VAR binding kết quả __enter__.
• Phần as VAR là optional. Có context manager không cần bind (vd lock).
• Khi block kết thúc (bình thường hoặc do exception), __exit__ được gọi.

Các use case phổ biến:

import threading

# File — close tự động
with open("a.txt", "w") as f:
    f.write("hi")

# Lock — release tự động
lock = threading.Lock()
with lock:
    shared_state += 1

# Database connection (SQLite) — commit/rollback + close
import sqlite3
with sqlite3.connect("db.sqlite") as conn:
    conn.execute("INSERT INTO ...")
    # commit khi không exception, rollback nếu có

10. Viết context manager dạng class

Để một object dùng được trong with, nó phải implement context manager protocol — 2 dunder method:

• __enter__(self) — chạy khi vào block. Return value được bind vào as VAR.
• __exit__(self, exc_type, exc_val, exc_tb) — chạy khi ra block. Nhận info về exception (nếu có).

import time

class Timer:
    def __init__(self, label="block"):
        self.label = label

    def __enter__(self):
        self.start = time.perf_counter()
        return self                # bind vào `as t`

    def __exit__(self, exc_type, exc_val, exc_tb):
        elapsed = time.perf_counter() - self.start
        print(f"{self.label}: {elapsed:.3f}s")
        # không return gì = return None = không suppress exception

with Timer("loop") as t:
    total = sum(range(10_000_000))

# loop: 0.142s

__exit__ arguments khi không có exception: cả 3 đều None.

__exit__ arguments khi có exception trong block: exc_type = class exception (vd ValueError), exc_val = instance exception, exc_tb = traceback object.

class IgnoreKeyError:
    def __enter__(self): return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is KeyError:
            print("Bỏ qua KeyError")
            return True      # suppress chỉ KeyError
        # return None ngầm cho exception khác → re-raise

with IgnoreKeyError():
    {}["missing"]    # raise KeyError, nhưng bị nuốt
print("vẫn chạy tiếp")

11. @contextlib.contextmanager — generator thành context manager

Viết class với __enter__/__exit__ đôi khi dài dòng cho case đơn giản. contextlib cung cấp một decorator biến generator function (yield đúng 1 lần) thành context manager. Cấu trúc try / yield / finally:

from contextlib import contextmanager
import time

@contextmanager
def timer(label="block"):
    start = time.perf_counter()
    try:
        yield start          # value bind vào `as`
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.3f}s")

with timer("loop"):
    total = sum(range(10_000_000))
# loop: 0.142s

Map ngữ nghĩa:

• Code trước yield = __enter__ (setup).
• Giá trị yield = giá trị trả về __enter__, bind vào as.
• Code trong finally = __exit__ (cleanup, luôn chạy).
• Code trong except (nếu có) = xử lý exception. Re-raise hoặc nuốt qua không re-raise.

Phiên bản đầy đủ với handle exception:

@contextmanager
def managed_resource(name):
    print(f"acquire {name}")
    res = acquire(name)
    try:
        yield res
    except ValueError as e:
        print(f"ValueError bị nuốt: {e}")
        # không re-raise = suppress exception
    finally:
        print(f"release {name}")
        release(res)

12. Nested with — nhiều context manager trong 1 dòng

Khi cần nhiều resource cùng lúc, viết liền nhau cách bởi dấu phẩy. Tránh nested deep indent:

# Cách viết phổ biến (Python 3.1+)
with open("in.txt") as fin, open("out.txt", "w") as fout:
    fout.write(fin.read().upper())

# Python 3.10+ cho phép parentheses để xuống dòng:
with (
    open("in.txt") as fin,
    open("out.txt", "w") as fout,
    lock,
):
    ...

Thứ tự: cm bên trái được __enter__ trước, cm bên phải sau. Khi exit: ngược lại (phải trước, trái sau) — đối xứng nested.

13. contextlib.ExitStack — số resource động

Khi số lượng resource chỉ biết tại runtime (vd: mở N file theo input), không thể viết tĩnh with open(a), open(b), .... Dùng ExitStack — collection-style context manager.

from contextlib import ExitStack

paths = ["a.txt", "b.txt", "c.txt"]   # động, chỉ biết runtime

with ExitStack() as stack:
    files = [stack.enter_context(open(p)) for p in paths]
    # tất cả file mở, cleanup khi ra block
    for f in files:
        process(f)
# Tất cả file được close, kể cả nếu giữa chừng có lỗi

API chính của ExitStack:

• enter_context(cm) — push 1 context manager vào stack, gọi __enter__, trả value.
• callback(fn, *args, **kwargs) — đăng ký 1 function gọi khi cleanup (giống defer của Go).
• push(cm) — push một cm đã enter từ trước.
• pop_all() — chuyển ownership cleanup sang ExitStack mới (advanced).

Ví dụ callback như defer:

with ExitStack() as stack:
    conn = db.connect()
    stack.callback(conn.close)        # defer close

    tmp = create_temp_file()
    stack.callback(os.remove, tmp)    # defer remove

    do_work(conn, tmp)
# Cleanup chạy theo LIFO: remove tmp trước, rồi close conn

14. contextlib.suppress — bỏ qua exception cụ thể

Idiom "thử làm gì đó, nếu lỗi loại X thì bỏ qua, lỗi khác thì re-raise" rất phổ biến. Cách "thủ công":

import os
try:
    os.remove("temp.txt")
except FileNotFoundError:
    pass

Cách Pythonic ngắn hơn với suppress:

from contextlib import suppress
import os

with suppress(FileNotFoundError):
    os.remove("temp.txt")

# Nhiều exception type cùng lúc:
with suppress(FileNotFoundError, PermissionError):
    os.remove("sensitive.log")

suppress chỉ "nuốt" đúng exception thuộc tuple đã chỉ định. Exception khác vẫn propagate bình thường — không phải catch-all except Exception (đó là anti-pattern).

Vài utility khác trong contextlib hay dùng:

Triết lý của suppress chính là dòng "unless explicitly silenced" trong Zen of Python: không catch bừa, nhưng cho phép tắt rõ ràng khi cần.

Bài tập

import functools, time

def measure_time(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return fn(*args, **kwargs)
        finally:
            elapsed = time.perf_counter() - start
            print(f"{fn.__name__}: {elapsed:.4f}s")
    return wrapper

@measure_time
def slow_sum(n):
    return sum(range(n))

slow_sum(10_000_000)  # slow_sum: 0.1421s

import functools, time

def retry(times=3, exceptions=(Exception,)):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kw):
            last_exc = None
            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kw)
                except exceptions as e:
                    last_exc = e
                    print(f"attempt {attempt} failed: {e}")
                    if attempt < times:
                        time.sleep(0.1 * attempt)
            raise last_exc
        return wrapper
    return decorator

@retry(times=3, exceptions=(IOError,))
def flaky():
    import random
    if random.random() < 0.8: raise IOError("network")
    return "ok"

import os, tempfile
from contextlib import contextmanager, suppress

@contextmanager
def temp_file(suffix=".txt"):
    fd, path = tempfile.mkstemp(suffix=suffix)
    os.close(fd)             # mkstemp mở sẵn, close để user tự open lại
    try:
        yield path
    finally:
        with suppress(FileNotFoundError):
            os.remove(path)

with temp_file() as p:
    with open(p, "w") as f:
        f.write("hello")
    print(open(p).read())
# File đã bị xoá sau block
assert not os.path.exists(p)

import functools, inspect, typing

def validate_types(fn):
    hints = typing.get_type_hints(fn)
    sig = inspect.signature(fn)

    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        bound = sig.bind(*args, **kwargs)
        bound.apply_defaults()
        for name, value in bound.arguments.items():
            expected = hints.get(name)
            if expected is None: continue
            if not isinstance(value, expected):
                raise TypeError(
                    f"{name}={value!r} không match {expected}"
                )
        return fn(*args, **kwargs)
    return wrapper

@validate_types
def greet(name: str, age: int) -> str:
    return f"{name} {age}"

greet("An", 25)        # OK
greet("An", "25")      # TypeError: age='25' không match <class 'int'>

def safe(fn):
    def wrapper(*args, **kw):
        try:
            return fn(*args, **kw)
        except ZeroDivisionError:
            return None
    return wrapper

@safe
def divide(a, b):
    """Chia a cho b. Trả None nếu chia 0."""
    return a / b

help(divide)  # Help on function wrapper... ← BUG
print(divide.__name__)  # 'wrapper' ← BUG

import functools

def safe(fn):
    @functools.wraps(fn)      # ← thêm dòng này
    def wrapper(*args, **kw):
        try:
            return fn(*args, **kw)
        except ZeroDivisionError:
            return None
    return wrapper

# Bây giờ:
print(divide.__name__)  # 'divide'
print(divide.__doc__)   # 'Chia a cho b. Trả None nếu chia 0.'

Quiz

Tổng kết

Sau Chương 7 bạn nên đã master:

• Decorator = function nhận function trả function. @dec def f = f = dec(f). Chạy at-import-time.
• Mọi decorator nên dùng @functools.wraps(fn) trên wrapper — bảo toàn metadata.
• Decorator có argument = factory: 3 tầng (factory → decorator → wrapper).
• Stack decorator: bottom-up apply, top-down execute. @a @b def f = a(b(f)).
• Memoization với @cache / @lru_cache — đặc biệt cho recursion. Args phải hashable.
• Class-based decorator: class + __call__. Tiện cho stateful decorator.
• Decorator stdlib/lib: @dataclass, @property, @route, @fixture… đâu cũng thấy.
• Context manager với with — RAII của Python, đảm bảo cleanup cả khi exception.
• 2 cách viết cm: class với __enter__/__exit__, hoặc generator với @contextmanager.
• __exit__ return True = suppress exception. Default = re-raise.
• Nhiều cm: viết liền dấu phẩy, hoặc ExitStack cho số động.
• suppress(ExceptionType) = idiom Pythonic cho "try / except / pass".

Kết nối

• Chương 3 (Functions & Closures) — wrapper trong decorator chính là closure capture fn. Hiểu closure giúp hiểu decorator.
• Chương 5 (OOP) — @property, @classmethod, @staticmethod, @dataclass đều là decorator. Nay bạn hiểu chúng vận hành thế nào.
• Chương 6 (Iterators & Generators) — @contextmanager biến generator (yield 1 lần) thành cm. Đây chính là use case quan trọng của generator ngoài lazy iteration.
• Chương 8 (Async Python) — Async có @asynccontextmanager và async with. Concept giống hệt, chỉ là async version.
• Chương 10 (Testing với pytest) — @pytest.fixture, @pytest.mark.parametrize là decorator. pytest fixture cũng dùng generator + yield giống @contextmanager.
• JS Chương 7 (HOC/closures) — JS không có decorator syntax (TC39 stage-3 đang chờ), nhưng pattern HOC withLogging(fn) tương đương Python decorator.