Context Managers Best Practice

Context Manager = Tự động dọn dẹp = Không bao giờ quên đóng file

Vấn đề: Quên đóng tài nguyên

# ❌ SAI: Quên đóng file
f = open("data.txt")
du_lieu = f.read()
# Lỗi xảy ra ở đây...
f.close()  # Không bao giờ chạy đến đây!

# ❌ TỐT HƠN: Dùng try-finally
f = open("data.txt")
try:
    du_lieu = f.read()
finally:
    f.close()  # Luôn đóng, nhưng code xấu

# ✅ TỐT NHẤT: Dùng Context Manager
with open("data.txt") as f:
    du_lieu = f.read()
# File tự động đóng khi ra khỏi block!

---

Context Manager là gì?

Context Manager là object quản lý setup và teardown tự động cho một block code.

Protocol

Bất kỳ object nào có 2 method sau là Context Manager:

class MyContextManager:
    def __enter__(self):
        """Chạy khi VÀO block with."""
        # Setup tài nguyên
        return self  # Giá trị gán cho biến sau "as"
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Chạy khi RA KHỎI block with (kể cả có lỗi)."""
        # Cleanup tài nguyên
        return False  # True = nuốt exception, False = để exception lan

---

Xây dựng Context Manager Đơn giản

Ví dụ: Timer Context Manager

import time

class Timer:
    """Đo thời gian thực thi của block code."""
    
    def __init__(self, ten: str = "Block"):
        self.ten = ten
        self.bat_dau = None
        self.ket_thuc = None
    
    def __enter__(self):
        self.bat_dau = time.perf_counter()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.ket_thuc = time.perf_counter()
        thoi_gian = self.ket_thuc - self.bat_dau
        print(f"⏱️ {self.ten}: {thoi_gian:.4f}s")
        return False  # Không nuốt exception

# Sử dụng
with Timer("Xử lý dữ liệu"):
    time.sleep(1.5)
    xu_ly_nang()
# Output: ⏱️ Xử lý dữ liệu: 1.5023s

Ví dụ: Database Connection

class DatabaseConnection:
    """Context manager cho kết nối database."""
    
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.connection = None
    
    def __enter__(self):
        print(f"Đang kết nối tới {self.connection_string}...")
        self.connection = self._tao_ket_noi()
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.connection:
            print("Đang đóng kết nối...")
            self.connection.close()
        
        # Nếu có lỗi, log lại
        if exc_type:
            print(f"Lỗi xảy ra: {exc_val}")
        
        return False  # Để exception lan ra ngoài
    
    def _tao_ket_noi(self):
        # Giả lập tạo kết nối
        return type("Connection", (), {"close": lambda self: None})()

# Sử dụng
with DatabaseConnection("postgresql://localhost/db") as conn:
    # Thực hiện query
    ket_qua = conn.execute("SELECT * FROM users")
# Kết nối tự động đóng!
)

---

Cách dễ hơn: @contextmanager

Decorator @contextmanager từ contextlib cho phép tạo context manager bằng generator.

from contextlib import contextmanager

@contextmanager
def timer(ten: str = "Block"):
    """Timer context manager dùng generator."""
    import time
    bat_dau = time.perf_counter()
    try:
        yield  # Code trong block with chạy ở đây
    finally:
        thoi_gian = time.perf_counter() - bat_dau
        print(f"⏱️ {ten}: {thoi_gian:.4f}s")

# Sử dụng - y hệt class!
with timer("Tính toán"):
    ket_qua = sum(range(1000000))

Cấu trúc @contextmanager

@contextmanager
def my_context():
    # Setup (tương đương __enter__)
    print("Bắt đầu")
    resource = create_resource()
    
    try:
        yield resource  # Trả về cho "as"
    finally:
        # Teardown (tương đương __exit__)
        cleanup(resource)
        print("Kết thúc")

---

Context Managers Có sẵn

File Handling

# Đọc file
with open("data.txt", "r", encoding="utf-8") as f:
    noi_dung = f.read()

# Ghi file (tự động flush và close)
with open("output.txt", "w") as f:
    f.write("Xin chào!")

# Đọc và ghi cùng lúc
with open("input.txt") as fin, open("output.txt", "w") as fout:
    fout.write(fin.read().upper())

Threading Lock

import threading

lock = threading.Lock()

# Với context manager
with lock:
    # Code này thread-safe
    shared_resource.modify()
# Lock tự động release

# Tương đương với:
lock.acquire()
try:
    shared_resource.modify()
finally:
    lock.release()

Suppress Exception

from contextlib import suppress

# Bỏ qua FileNotFoundError
with suppress(FileNotFoundError):
    os.remove("file_khong_ton_tai.txt")
# Không lỗi nếu file không tồn tại!

Temporary Directory

import tempfile

with tempfile.TemporaryDirectory() as tmpdir:
    # tmpdir là đường dẫn tới thư mục tạm
    print(f"Thư mục tạm: {tmpdir}")
    # Tạo file, xử lý, v.v.
# Thư mục tự động xóa khi ra khỏi block!

Change Directory

import os
from contextlib import contextmanager

@contextmanager
def chuyen_thu_muc(path: str):
    """Tạm thời chuyển sang thư mục khác."""
    thu_muc_hien_tai = os.getcwd()
    try:
        os.chdir(path)
        yield
    finally:
        os.chdir(thu_muc_hien_tai)

with chuyen_thu_muc("/tmp"):
    # Làm việc trong /tmp
    print(os.getcwd())  # /tmp
print(os.getcwd())  # Quay lại thư mục cũ

---

Xử lý Exception trong __exit__

__exit__ nhận 3 tham số về exception:

def __exit__(self, exc_type, exc_val, exc_tb):
    """
    exc_type: Kiểu exception (None nếu không có lỗi)
    exc_val: Giá trị exception
    exc_tb: Traceback object
    """
    if exc_type is None:
        print("Không có lỗi")
    else:
        print(f"Có lỗi: {exc_type.__name__}: {exc_val}")
    
    # Return True để "nuốt" exception
    # Return False để exception lan ra ngoài
    return False

Ví dụ: Transaction Rollback

class Transaction:
    def __init__(self, db):
        self.db = db
    
    def __enter__(self):
        self.db.begin()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            # Có lỗi → Rollback
            print("Rolling back...")
            self.db.rollback()
        else:
            # Không lỗi → Commit
            print("Committing...")
            self.db.commit()
        return False

# Sử dụng
with Transaction(db) as tx:
    db.insert({"ten": "HPN"})
    db.update({"tuoi": 28})
    # Nếu có lỗi → tự động rollback
    # Không lỗi → tự động commit

---

Nhiều Context Managers

# Cách 1: Lồng nhau
with open("input.txt") as fin:
    with open("output.txt", "w") as fout:
        fout.write(fin.read())

# Cách 2: Trên cùng một dòng
with open("input.txt") as fin, open("output.txt", "w") as fout:
    fout.write(fin.read())

# Cách 3: Xuống dòng với dấu ngoặc (Python 3.10+)
with (
    open("input.txt") as fin,
    open("output.txt", "w") as fout,
    Timer("Copy file") as t,
):
    fout.write(fin.read())

---

Async Context Manager

Cho code bất đồng bộ:

class AsyncDatabaseConnection:
    async def __aenter__(self):
        self.conn = await create_connection()
        return self.conn
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()

# Sử dụng
async with AsyncDatabaseConnection() as conn:
    await conn.execute("SELECT * FROM users")

---

Bảng Tóm tắt

# === CLASS-BASED CONTEXT MANAGER ===
class MyContext:
    def __enter__(self):
        # Setup
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Cleanup
        return False  # Không nuốt exception

# === GENERATOR-BASED CONTEXT MANAGER ===
from contextlib import contextmanager

@contextmanager
def my_context():
    # Setup
    try:
        yield resource
    finally:
        # Cleanup

# === BUILT-IN CONTEXT MANAGERS ===
with open(path) as f:                    # File
with lock:                               # Threading Lock
with suppress(Exception):                # Bỏ qua exception
with TemporaryDirectory() as d:          # Thư mục tạm

# === MULTIPLE CONTEXT MANAGERS ===
with cm1() as a, cm2() as b:
    ...

# === ASYNC CONTEXT MANAGER ===
async with async_cm() as resource:
    ...