PENALGO

Giao diện

Context Managers — Quản lý tài nguyên không bao giờ rò rỉ ​

Đội ops nhận alert lúc 2 giờ sáng — database connection pool exhausted. 500 connections đang mở, không connection nào được đóng đúng cách. Service trả về HTTP 503 liên tục, khách hàng mất khả năng truy cập trong 47 phút. Root cause? Một hàm xử lý request quên gọi connection.close() khi exception xảy ra giữa chừng. Chỉ cần một dòng code thiếu, toàn bộ hệ thống sụp đổ.

Resource leak là một trong những nguyên nhân gây sự cố production phổ biến nhất — từ file descriptor cạn kiệt, database connection pool tràn, đến lock không được giải phóng gây deadlock. Vấn đề không nằm ở việc lập trình viên "quên" cleanup, mà ở việc cleanup code bị bỏ qua khi exception xảy ra trên đường thực thi bình thường. try/finally giải quyết được về mặt kỹ thuật, nhưng code trở nên lồng nhau phức tạp và dễ sai khi có nhiều tài nguyên cần quản lý đồng thời.

Context manager là câu trả lời của Python cho bài toán này. Thông qua câu lệnh with, Python đảm bảo rằng mọi tài nguyên được acquire sẽ luôn luôn được release — dù code bên trong chạy thành công, raise exception, hay thậm chí bị return giữa chừng. Đây không chỉ là syntactic sugar; đây là một protocol được tích hợp sâu vào bytecode interpreter, biến resource management từ "trách nhiệm của lập trình viên" thành "đảm bảo của ngôn ngữ".

Bức tranh tư duy ​

Hãy hình dung bạn đặt phòng họp tại một tòa nhà văn phòng chuyên nghiệp. Mỗi phòng họp đều có một lễ tân riêng. Khi bạn đến, lễ tân mở cửa phòng, bật đèn, bật máy lạnh, chuẩn bị bảng trắng và bút — tất cả setup cần thiết để bạn làm việc hiệu quả. Đó chính là __enter__.

Khi bạn ra khỏi phòng — dù cuộc họp kết thúc đúng giờ, kết thúc sớm, hay thậm chí bạn phải rời đi giữa chừng vì có sự cố khẩn cấp — lễ tân luôn luôn tắt đèn, tắt máy lạnh, lau bảng, và khóa cửa. Đó là __exit__. Lễ tân không quan tâm lý do bạn rời đi; nhiệm vụ cleanup luôn được thực hiện.

Context manager chính là "lễ tân" cho tài nguyên trong chương trình. Bạn không cần nhớ phải đóng file, release lock, hay close connection — context manager đảm bảo điều đó xảy ra tự động, kể cả khi exception xảy ra.

┌─────────────────────────────────────────────┐
│              with resource() as r:          │
│  ┌─────────┐                                │
│  │ ENTER   │  Mở cửa, bật đèn, setup       │
│  └────┬────┘                                │
│       ▼                                     │
│  ┌─────────┐                                │
│  │  BODY   │  Bạn làm việc trong phòng họp  │
│  └────┬────┘                                │
│       ▼         (dù có exception hay không) │
│  ┌─────────┐                                │
│  │  EXIT   │  Tắt đèn, khóa cửa, cleanup   │
│  └─────────┘                                │
└─────────────────────────────────────────────┘

Cốt lõi kỹ thuật ​

Protocol: __enter__ và __exit__ ​

Context Manager Protocol yêu cầu object implement hai dunder method. Khi Python gặp câu lệnh with, nó gọi __enter__ trước khi thực thi body, và đảm bảo gọi __exit__ sau khi body kết thúc — bất kể kết thúc bình thường hay do exception.

python
class ManagedFile:
    """Context manager cho file operations với logging."""

    def __init__(self, filename: str, mode: str = "r") -> None:
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        """Được gọi khi vào block with. Return value gán cho biến sau 'as'."""
        self.file = open(self.filename, self.mode, encoding="utf-8")
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        """Luôn được gọi khi rời block with.

        Parameters:
            exc_type: Class của exception (None nếu không có lỗi)
            exc_val:  Instance của exception
            exc_tb:   Traceback object

        Returns:
            True  → nuốt exception, code tiếp tục chạy bình thường
            False → re-raise exception sau khi cleanup
        """
        if self.file:
            self.file.close()
        # Luôn trả False để không nuốt exception
        return False


# Sử dụng
with ManagedFile("config.json", "r") as f:
    data = f.read()
# f.close() đã được gọi tự động — kể cả khi f.read() raise exception

Điểm quan trọng: giá trị return của __exit__ quyết định exception có bị "nuốt" hay không. Trả về True nghĩa là exception đã được xử lý và không cần propagate. Trả về False (hoặc None) nghĩa là exception sẽ tiếp tục lan ra ngoài sau khi cleanup hoàn tất.

@contextmanager từ contextlib ​

Viết một class đầy đủ với __enter__/__exit__ cho những context manager đơn giản thường là overkill. Module contextlib cung cấp decorator @contextmanager cho phép viết context manager dưới dạng generator function — code trước yield là setup, code sau yield là teardown.

python
from contextlib import contextmanager
import time


@contextmanager
def execution_timer(label: str = "Block"):
    """Đo và in thời gian thực thi của code block."""
    start = time.perf_counter()
    try:
        yield  # Control chuyển cho body của with statement
    finally:
        elapsed = time.perf_counter() - start
        print(f"⏱️ {label}: {elapsed:.4f}s")


with execution_timer("Data processing"):
    total = sum(range(10_000_000))
# Output: ⏱️ Data processing: 0.3172s

Cấu trúc bắt buộc: try/finally bao quanh yield là không thể thiếu. Nếu thiếu try/finally, cleanup code sẽ không chạy khi exception xảy ra trong body — đây là một trong những sai lầm phổ biến nhất.

python
@contextmanager
def database_transaction(connection):
    """Transaction manager với commit/rollback tự động."""
    tx = connection.begin()
    try:
        yield tx
    except Exception:
        tx.rollback()
        raise  # Re-raise sau khi rollback
    else:
        tx.commit()  # Chỉ commit nếu không có exception

contextlib utilities ​

Module contextlib cung cấp nhiều context manager tiện ích cho các pattern thường gặp:

python
from contextlib import suppress, redirect_stdout, closing, ExitStack
import io


# suppress — bỏ qua exception cụ thể
with suppress(FileNotFoundError):
    os.remove("/tmp/obsolete_cache.dat")
# Tương đương try/except FileNotFoundError: pass — nhưng rõ ràng hơn


# redirect_stdout — chuyển hướng stdout tạm thời
buffer = io.StringIO()
with redirect_stdout(buffer):
    print("Captured output")
captured = buffer.getvalue()  # "Captured output\n"


# closing — wrap object có .close() thành context manager
from urllib.request import urlopen

with closing(urlopen("https://api.example.com/data")) as response:
    data = response.read()
# response.close() được gọi tự động


# ExitStack — quản lý số lượng context manager động
def process_files(filenames: list[str]) -> list[str]:
    """Mở nhiều file cùng lúc, đảm bảo tất cả đều được đóng."""
    with ExitStack() as stack:
        files = [
            stack.enter_context(open(fname, encoding="utf-8"))
            for fname in filenames
        ]
        return [f.read() for f in files]
    # Tất cả file đều được đóng khi rời ExitStack

ExitStack đặc biệt mạnh khi số lượng context manager không biết trước — ví dụ mở N file từ list, hoặc acquire N lock dựa trên runtime conditions.

Multiple context managers ​

Python hỗ trợ nhiều cách sử dụng nhiều context manager cùng lúc:

python
# Cách 1: Trên cùng một dòng (ngắn gọn cho 2-3 managers)
with open("input.csv") as fin, open("output.csv", "w") as fout:
    fout.write(fin.read())

# Cách 2: Parenthesized syntax (Python 3.10+) — đọc dễ hơn
with (
    open("source.log", encoding="utf-8") as source,
    open("filtered.log", "w", encoding="utf-8") as dest,
    execution_timer("Log filtering"),
):
    for line in source:
        if "ERROR" in line:
            dest.write(line)

# Cách 3: ExitStack cho số lượng động
def merge_sorted_files(paths: list[str], output: str) -> None:
    with ExitStack() as stack:
        files = [
            stack.enter_context(open(p, encoding="utf-8"))
            for p in paths
        ]
        out = stack.enter_context(open(output, "w", encoding="utf-8"))
        # merge logic...

Async context managers: __aenter__ / __aexit__ ​

Trong môi trường async, tài nguyên thường yêu cầu I/O không đồng bộ để acquire và release. Python cung cấp async context manager protocol với __aenter__/__aexit__, sử dụng qua async with.

python
import asyncio
from contextlib import asynccontextmanager


class AsyncDBPool:
    """Async context manager cho database connection pool."""

    def __init__(self, dsn: str, min_size: int = 5, max_size: int = 20) -> None:
        self.dsn = dsn
        self.min_size = min_size
        self.max_size = max_size
        self.pool = None

    async def __aenter__(self):
        self.pool = await asyncpg.create_pool(
            self.dsn, min_size=self.min_size, max_size=self.max_size
        )
        return self.pool

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> bool:
        if self.pool:
            await self.pool.close()
        return False


# Sử dụng
async def main():
    async with AsyncDBPool("postgresql://localhost/app") as pool:
        async with pool.acquire() as conn:
            rows = await conn.fetch("SELECT id, name FROM users LIMIT 10")


# Generator-based async context manager
@asynccontextmanager
async def managed_http_session():
    """Quản lý aiohttp session lifecycle."""
    import aiohttp

    session = aiohttp.ClientSession()
    try:
        yield session
    finally:
        await session.close()


async def fetch_data():
    async with managed_http_session() as session:
        async with session.get("https://api.example.com/data") as resp:
            return await resp.json()

Thực chiến ​

Production scenario: Database Transaction Manager cho FastAPI ​

Trong production, transaction management cần xử lý commit/rollback, savepoints, logging, và pool monitoring. Dưới đây là implementation sử dụng SQLAlchemy.

python
from __future__ import annotations

import logging
import time
from contextlib import contextmanager
from typing import Generator

from sqlalchemy import event, text
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session, sessionmaker

logger = logging.getLogger(__name__)


class TransactionManager:
    """Production-grade transaction manager với monitoring và savepoint support."""

    def __init__(self, session_factory: sessionmaker) -> None:
        self._session_factory = session_factory
        self._active_transactions: int = 0

    @contextmanager
    def transaction(self, read_only: bool = False) -> Generator[Session, None, None]:
        """Context manager cho database transaction."""
        session: Session = self._session_factory()
        self._active_transactions += 1
        tx_start = time.monotonic()

        try:
            if read_only:
                session.execute(text("SET TRANSACTION READ ONLY"))

            yield session

            session.commit()
            elapsed = time.monotonic() - tx_start
            logger.info("Transaction committed (%.3fs)", elapsed)

        except Exception as exc:
            elapsed = time.monotonic() - tx_start
            session.rollback()
            logger.error(
                "Transaction rolled back after %.3fs: %s", elapsed, exc
            )
            raise

        finally:
            self._active_transactions -= 1
            session.close()

    @contextmanager
    def savepoint(self, session: Session, name: str) -> Generator[None, None, None]:
        """Nested transaction sử dụng SAVEPOINT."""
        nested = session.begin_nested()
        try:
            yield
            nested.commit()
        except Exception as exc:
            nested.rollback()
            logger.warning("Savepoint '%s' rolled back: %s", name, exc)
            raise

    @property
    def active_count(self) -> int:
        return self._active_transactions


# ---------- Tích hợp với FastAPI ----------

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

engine = Engine  # placeholder — thay bằng create_engine() thực tế
SessionLocal = sessionmaker(bind=engine)
tx_manager = TransactionManager(SessionLocal)


def get_tx() -> TransactionManager:
    return tx_manager


class OrderCreate(BaseModel):
    user_id: int
    product_id: int
    quantity: int


@app.post("/orders")
def create_order(
    order: OrderCreate,
    tx: TransactionManager = Depends(get_tx),
):
    with tx.transaction() as session:
        # Kiểm tra inventory trong savepoint riêng
        with tx.savepoint(session, "check_inventory"):
            product = session.execute(
                text("SELECT stock FROM products WHERE id = :pid"),
                {"pid": order.product_id},
            ).scalar_one_or_none()

            if product is None or product < order.quantity:
                raise HTTPException(status_code=400, detail="Không đủ hàng")

        # Tạo order và trừ inventory
        session.execute(
            text(
                "INSERT INTO orders (user_id, product_id, quantity) "
                "VALUES (:uid, :pid, :qty)"
            ),
            {
                "uid": order.user_id,
                "pid": order.product_id,
                "qty": order.quantity,
            },
        )
        session.execute(
            text("UPDATE products SET stock = stock - :qty WHERE id = :pid"),
            {"qty": order.quantity, "pid": order.product_id},
        )

    return {"status": "created"}
    # Nếu bất kỳ operation nào fail → toàn bộ transaction rollback
    # Connection trả về pool → không bao giờ leak

Pattern này đảm bảo ba thuộc tính quan trọng: atomicity (toàn bộ transaction thành công hoặc rollback), cleanup (session luôn được đóng trong finally), và observability (mọi transaction đều được log với thời gian thực thi).

Sai lầm điển hình ​

Sai lầm 1: __exit__ return True nuốt exception ​

python
# ❌ SAI: Return True nuốt mọi exception — silent failures
class BadManager:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cleanup()
        return True  # Nuốt TẤT CẢ exception!


with BadManager():
    raise ValueError("Dữ liệu không hợp lệ")
# Không có exception nào được raise — bug bị che giấu hoàn toàn
# Production impact: lỗi logic chạy im lặng, data corruption không ai biết
python
# ✅ ĐÚNG: Chỉ return True khi cố ý xử lý exception cụ thể
class GoodManager:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cleanup()
        # Chỉ suppress exception cụ thể đã được xử lý
        if exc_type is ConnectionResetError:
            logger.warning("Connection reset — đã xử lý gracefully")
            return True
        return False  # Mọi exception khác tiếp tục propagate

Sai lầm 2: Thiếu try/finally trong @contextmanager ​

python
# ❌ SAI: Không có try/finally — resource leak khi exception xảy ra
@contextmanager
def open_connection(dsn: str):
    conn = psycopg2.connect(dsn)
    yield conn
    conn.close()  # KHÔNG BAO GIỜ CHẠY nếu body raise exception!
# Production impact: connection pool cạn kiệt dần, service degradation
python
# ✅ ĐÚNG: try/finally đảm bảo cleanup luôn xảy ra
@contextmanager
def open_connection(dsn: str):
    conn = psycopg2.connect(dsn)
    try:
        yield conn
    finally:
        conn.close()  # Luôn chạy — dù có exception hay không

Sai lầm 3: Giữ reference sau khi context manager exit ​

python
# ❌ SAI: Sử dụng resource sau khi with block kết thúc
with open("data.txt") as f:
    header = f.readline()

# f đã đóng nhưng biến vẫn tồn tại trong scope
remaining = f.read()  # ValueError: I/O operation on closed file

# Production impact: intermittent errors khó debug vì biến vẫn accessible
python
# ✅ ĐÚNG: Trích xuất dữ liệu cần thiết bên trong with block
with open("data.txt") as f:
    header = f.readline()
    content = f.read()  # Đọc hết trong khi file còn mở

# Sử dụng data đã trích xuất — không cần file handle nữa
process(header, content)

Sai lầm 4: Không handle exception trong __exit__ properly ​

python
# ❌ SAI: __exit__ tự raise exception — che mất exception gốc
class LeakyManager:
    def __enter__(self):
        self.resource = acquire_resource()
        return self.resource

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.resource.close()  # Nếu close() raise → exception gốc bị mất!
        return False


# Nếu body raise ValueError và close() raise IOError:
# Chỉ IOError được propagate — ValueError biến mất hoàn toàn
# Production impact: root cause bị mất, debugging trở nên cực kỳ khó
python
# ✅ ĐÚNG: Bảo vệ cleanup code, giữ nguyên exception gốc
class SafeManager:
    def __enter__(self):
        self.resource = acquire_resource()
        return self.resource

    def __exit__(self, exc_type, exc_val, exc_tb):
        try:
            self.resource.close()
        except Exception:
            # Log lỗi cleanup nhưng không che mất exception gốc
            logger.exception("Cleanup failed for resource")
            if exc_type is None:
                raise  # Chỉ raise cleanup error nếu không có exception gốc
        return False

Under the Hood ​

Bytecode: with statement compiled ​

Khi CPython biên dịch câu lệnh with, nó sinh ra các bytecode instruction đặc biệt để đảm bảo __exit__ luôn được gọi. Sử dụng module dis để quan sát:

python
import dis

def example():
    with open("test.txt") as f:
        data = f.read()

dis.dis(example)

Simplified bytecode flow (CPython 3.12+):

LOAD_GLOBAL          open
LOAD_CONST           'test.txt'
CALL                 1
BEFORE_WITH                      # Push __exit__, gọi __enter__
STORE_FAST           f
# --- Body ---
LOAD_FAST            f
LOAD_ATTR            read
CALL                 0
STORE_FAST           data
# --- Cleanup (no exception) ---
LOAD_CONST           None        # exc_type, exc_val, exc_tb = None
CALL                 3           # __exit__(None, None, None)

Khi exception xảy ra trong body, CPython tự động đặt exception info lên stack và gọi __exit__(exc_type, exc_val, exc_tb). Nếu __exit__ trả về truthy value, exception bị suppress.

ExitStack internals: LIFO ordering ​

ExitStack lưu các cleanup callback trong một list và gọi chúng theo thứ tự LIFO (Last In, First Out) — giống với cách nested with statements hoạt động:

python
from contextlib import ExitStack

with ExitStack() as stack:
    stack.enter_context(open("file_a.txt"))   # Đóng cuối cùng
    stack.enter_context(open("file_b.txt"))   # Đóng thứ hai
    stack.callback(print, "Cleanup custom")    # Chạy đầu tiên

# Thứ tự cleanup:
# 1. print("Cleanup custom")
# 2. file_b.close()
# 3. file_a.close()

Thứ tự LIFO là thiết kế có chủ đích: tài nguyên acquire sau thường phụ thuộc vào tài nguyên acquire trước, nên cần release trước (đóng cursor trước connection, giải phóng child lock trước parent lock).

Performance: class-based vs generator-based ​

Generator-based context manager (@contextmanager) có overhead nhỏ hơn class-based vì tránh tạo instance object. Tuy nhiên, sự khác biệt chỉ đáng kể khi context manager được gọi hàng triệu lần trong tight loop:

python
# Benchmark (1 triệu iterations)
# Class-based:     ~0.45s — tạo instance + method lookup
# Generator-based: ~0.38s — frame creation nhẹ hơn
# Overhead so với bare code: ~200ns/call cho cả hai

Trong production, overhead 200ns/call là không đáng kể so với I/O operation (database query: ~1-50ms, file read: ~0.1-10ms). Chọn approach dựa trên readability, không phải performance.

Async context managers và event loop ​

Async context managers (__aenter__/__aexit__) hoạt động tương tự synchronous version, nhưng các method là coroutines được schedule bởi event loop:

python
# async with KHÔNG block event loop trong khi chờ acquire/release
async with aiohttp.ClientSession() as session:
    # __aenter__: event loop chạy task khác trong khi chờ TCP connect
    async with session.get(url) as response:
        data = await response.json()
    # __aexit__: event loop chạy task khác trong khi chờ connection close

Bytecode cho async with sử dụng BEFORE_ASYNC_WITH và GET_AWAITABLE thay vì BEFORE_WITH, cho phép interpreter await kết quả của __aenter__ và __aexit__.

Checklist ghi nhớ ​

✅ Checklist triển khai

  • [ ] Mọi tài nguyên (file, connection, lock) phải được quản lý bằng with statement
  • [ ] __exit__ chỉ return True khi cố ý suppress exception cụ thể — mặc định luôn return False
  • [ ] @contextmanager bắt buộc có try/finally bao quanh yield để đảm bảo cleanup
  • [ ] Không sử dụng resource reference sau khi with block kết thúc — trích xuất data bên trong block
  • [ ] Cleanup code trong __exit__ phải được bảo vệ bằng try/except riêng để không che mất exception gốc
  • [ ] Sử dụng ExitStack khi số lượng context manager không biết trước tại compile time
  • [ ] Async resources sử dụng async with và __aenter__/__aexit__ — không dùng sync context manager trong async code
  • [ ] Transaction manager phải handle cả commit (happy path) và rollback (exception path) rõ ràng
  • [ ] Thứ tự cleanup là LIFO — tài nguyên acquire sau được release trước
  • [ ] Test context manager với cả happy path và exception path — đặc biệt verify cleanup xảy ra khi raise
  • [ ] Logging trong __enter__/__exit__ giúp debug resource lifecycle trong production
  • [ ] Ưu tiên @contextmanager cho logic đơn giản; dùng class-based khi cần lưu state phức tạp hoặc reuse
  • [ ] Python 3.10+ parenthesized with syntax giúp code đọc dễ hơn khi có nhiều manager

Bài tập luyện tập ​

Bài 1 — Foundation: Timer Context Manager ​

Yêu cầu: Viết class Timer hoạt động như context manager, đo thời gian thực thi. Timer phải hỗ trợ:

  • Đo thời gian bằng time.perf_counter() (precision cao)
  • In kết quả khi exit với format [Timer: {label}] {elapsed:.4f}s
  • Lưu elapsed attribute để truy cập sau khi exit
  • Không nuốt exception
python
# Expected usage:
with Timer("Sort algorithm") as t:
    sorted(range(1_000_000), reverse=True)
# Output: [Timer: Sort algorithm] 0.1234s

print(f"Elapsed: {t.elapsed}")  # Truy cập sau khi exit

🧠 Quiz — Timer Context Manager

Khi exception xảy ra trong body của with Timer("test") as t, điều gì xảy ra?

A. Timer không in thời gian và exception propagate B. Timer in thời gian, t.elapsed được set, và exception propagate C. Timer in thời gian và exception bị nuốt D. Timer raise RuntimeError thay vì exception gốc

Đáp án: B

__exit__ luôn được gọi khi rời with block — kể cả khi có exception. Vì __exit__ return False, exception tiếp tục propagate sau khi Timer hoàn tất công việc cleanup (tính elapsed và print). Attribute t.elapsed vẫn accessible vì object Timer tồn tại ngoài with block scope.

Lời giải Bài 1
python
import time


class Timer:
    """Context manager đo thời gian thực thi."""

    def __init__(self, label: str = "Block") -> None:
        self.label = label
        self.elapsed: float = 0.0
        self._start: float = 0.0

    def __enter__(self) -> "Timer":
        self._start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        self.elapsed = time.perf_counter() - self._start
        print(f"[Timer: {self.label}] {self.elapsed:.4f}s")
        return False  # Không nuốt exception


    # Verification
    # Happy path
    with Timer("Sum") as t:
        total = sum(range(10_000_000))
    print(f"Elapsed: {t.elapsed:.4f}s")

    # Exception path
    try:
        with Timer("Will fail") as t2:
            raise ValueError("Test error")
    except ValueError:
        print(f"Timer recorded: {t2.elapsed:.4f}s")

Bài 2 — Intermediate: TempDatabase Context Manager ​

Yêu cầu: Viết context manager TempDatabase sử dụng @contextmanager cho testing. Manager phải:

  • Tạo một SQLite database tạm thời (sử dụng tempfile.NamedTemporaryFile)
  • Tạo schema cơ bản (bảng users với id, name, email)
  • Yield connection object để test code sử dụng
  • Khi exit: đóng connection và xóa file database
  • Đảm bảo cleanup xảy ra kể cả khi test raise exception
python
# Expected usage:
with TempDatabase() as conn:
    conn.execute("INSERT INTO users (name, email) VALUES (?, ?)",
                 ("Alice", "alice@example.com"))
    conn.commit()
    row = conn.execute("SELECT * FROM users").fetchone()
    assert row[1] == "Alice"
# Database file đã bị xóa hoàn toàn

🧠 Quiz — TempDatabase cleanup

Nếu bạn quên try/finally trong @contextmanager và test code raise exception, điều gì xảy ra?

A. Python tự động cleanup vì garbage collector B. Database file tồn tại vĩnh viễn trên disk, connection không được đóng C. @contextmanager tự động thêm try/finally D. Exception bị nuốt và test pass giả

Đáp án: B

Không có try/finally, code sau yield không chạy khi exception xảy ra. Connection không được đóng (resource leak) và temp file không được xóa (disk space leak). Garbage collector có thể eventually đóng connection khi object bị thu hồi, nhưng thời điểm không xác định và file sẽ không bị xóa. Đây là lý do try/finally là bắt buộc trong @contextmanager.

Lời giải Bài 2
python
import os
import sqlite3
import tempfile
from contextlib import contextmanager


SCHEMA = """
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    email TEXT NOT NULL UNIQUE
);
"""


@contextmanager
def TempDatabase():
    """Context manager tạo SQLite database tạm cho testing.

    Yields:
        sqlite3.Connection đã có schema, sẵn sàng sử dụng.
    """
    # Tạo temp file — delete=False vì SQLite cần path ổn định
    tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
    db_path = tmp.name
    tmp.close()  # Đóng file handle để SQLite có thể mở

    conn = sqlite3.connect(db_path)
    try:
        conn.execute(SCHEMA)
        conn.commit()
        yield conn
    finally:
        # Cleanup: đóng connection trước, xóa file sau
        try:
            conn.close()
        except Exception:
            pass  # Connection có thể đã đóng
        try:
            os.unlink(db_path)
        except OSError:
            pass  # File có thể đã bị xóa


    # Verification
    with TempDatabase() as conn:
        conn.execute(
            "INSERT INTO users (name, email) VALUES (?, ?)",
            ("Alice", "alice@example.com"),
        )
        conn.commit()
        row = conn.execute("SELECT * FROM users").fetchone()
        assert row[1] == "Alice"
        print(f"User: {row[1]}, Email: {row[2]}")

    # Exception path — cleanup vẫn xảy ra
    try:
        with TempDatabase() as conn:
            conn.execute(
                "INSERT INTO users (name, email) VALUES (?, ?)",
                ("Bob", "bob@example.com"),
            )
            raise RuntimeError("Simulated test failure")
    except RuntimeError:
        print("Database file đã được cleanup")

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

📚 Bài liên quan

Glossary keywords: context manager, with statement, __enter__, __exit__, contextlib, @contextmanager, ExitStack, suppress, async context manager, __aenter__, __aexit__, @asynccontextmanager, LIFO cleanup, resource management, transaction manager, savepoint, connection pool.

Professional Engineering Standards | © 2025 HPN Corporation