Giao diện
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
@contextmanagerdecorator 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ọitime.perf_counter()và lưu vàoself._start__enter__phải returnselfđể sử dụngas t__exit__nhận 3 tham số exception, returnFalseđể re-raise exceptions
Gợi ý Bài 2
- Pattern
@contextmanager: code trướcyield= setup, code sauyield= 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__và__aexit__là phiên bản async, dùngasync with- Rollback transaction nếu
exc_type is not None - Dùng
asynccontextmanagercho 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)