Skip to content

Thực hành: Context Manager Patterns

🎯 Mục tiêu

🎯 Sau bài thực hành này, bạn sẽ:

  • Triển khai context manager bằng class (__enter__/__exit__)
  • Dùng @contextmanager decorator cho trường hợp đơn giản
  • Xây dựng async context manager cho tài nguyên bất đồng bộ

Mô tả bài tập

Context manager đảm bảo tài nguyên (file, connection, lock) luôn được giải phóng đúng cách, ngay cả khi có exception. Đây là pattern quan trọng trong Python professional development.

Yêu cầu

Bài 1: Class-based Context Manager

Tạo Timer context manager đo thời gian thực thi block code.

python
import time

class Timer:
    """Context manager đo thời gian thực thi."""
    def __init__(self, label: str = "Block"):
        self.label = label
        self.elapsed = 0.0
    def __enter__(self):
        # TODO: Ghi nhận thời điểm bắt đầu
        # Return self để dùng trong 'as'
        pass
    def __exit__(self, exc_type, exc_val, exc_tb):
        # TODO: Tính elapsed time và in ra
        # Return False để không suppress exceptions
        pass
# Sử dụng
with Timer("Sorting") as t:
    sorted(range(100000, 0, -1))
print(f"Mất {t.elapsed:.4f}s")

Bài 2: @contextmanager Decorator

Dùng contextlib.contextmanager để viết context manager ngắn gọn hơn.

python
from contextlib import contextmanager
@contextmanager
def temporary_directory():
    """Tạo thư mục tạm, tự xóa khi thoát context."""
    import tempfile
    import shutil
    # TODO: Tạo temp dir
    # yield path
    # Cleanup trong finally block
    pass

@contextmanager
def suppress_errors(*exception_types):
    """Bắt và log errors thay vì crash."""
    # TODO: try/yield/except — log error nhưng không raise
    pass
# Test
with suppress_errors(ValueError, TypeError):
    int("not_a_number")  # Không crash
print("Chương trình vẫn chạy!")

Bài 3: Nested và Async Context Manager

Quản lý nhiều tài nguyên đồng thời và xây dựng async context manager.

python
from contextlib import asynccontextmanager
class DatabaseConnection:
    """Giả lập database connection với context manager."""
    def __init__(self, db_url: str):
        self.db_url = db_url
        self.connected = False
    async def __aenter__(self):
        # TODO: Giả lập kết nối async
        pass
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # TODO: Đóng kết nối, rollback nếu có exception
        pass
# Sử dụng
# async with DatabaseConnection("postgresql://...") as db:
#     await db.execute("SELECT * FROM users")

Gợi ý

Gợi ý Bài 1
  • __enter__ gọi time.perf_counter() và lưu vào self._start
  • __enter__ phải return self để sử dụng as t
  • __exit__ nhận 3 tham số exception, return False để re-raise exceptions
Gợi ý Bài 2
  • Pattern @contextmanager: code trước yield = setup, code sau yield = cleanup
  • Luôn dùng try/finally để đảm bảo cleanup chạy ngay cả khi có exception
  • suppress_errors: try: yield / except exception_types as e: print(e)
Gợi ý Bài 3
  • __aenter____aexit__ là phiên bản async, dùng async with
  • Rollback transaction nếu exc_type is not None
  • Dùng asynccontextmanager cho version decorator

Lời giải tham khảo

Xem lời giải
python
import time
from contextlib import contextmanager

class Timer:
    def __init__(self, label: str = "Block"):
        self.label = label
        self.elapsed = 0.0
    def __enter__(self):
        self._start = time.perf_counter()
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.perf_counter() - self._start
        print(f"⏱️ {self.label}: {self.elapsed:.4f}s")
        return False
@contextmanager
def temporary_directory():
    import tempfile, shutil
    path = tempfile.mkdtemp()
    try:
        yield path
    finally:
        shutil.rmtree(path, ignore_errors=True)