Context Managers — with Statements and Resource Management

A context manager is any object that defines __enter__ and __exit__ methods, making it usable with the with statement. The with statement guarantees that __exit__ is called when the block ends — whether normally, by return, or by exception. This guarantee makes context managers the correct tool for any resource that must be released after use: file handles, database connections, network sockets, thread locks, and temporary directories. In FastAPI, database sessions are managed as context managers to ensure every session is closed — even when a request raises an exception.

The with Statement #

# File — the most common context manager
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()
# f is automatically closed here — even if f.read() raises an exception

# Multiple context managers in one with (Python 3.10+: parenthesised)
with (
    open("input.txt",  "r", encoding="utf-8") as infile,
    open("output.txt", "w", encoding="utf-8") as outfile,
):
    for line in infile:
        outfile.write(line.upper())

# Older Python (all versions):
with open("a.txt") as a, open("b.txt") as b:
    # both files open in this block
    pass

# Lock — ensures lock is always released
import threading
lock = threading.Lock()
with lock:
    shared_counter += 1
# lock.release() called automatically even if exception occurs inside

Building Context Managers with Classes #

class DatabaseSession:
    """Context manager that provides a database session and handles commit/rollback."""

    def __init__(self, db_engine):
        self.engine  = db_engine
        self.session = None

    def __enter__(self):
        """Called at start of 'with' block."""
        from sqlalchemy.orm import Session
        self.session = Session(self.engine)
        return self.session   # bound to 'as' variable

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Called at end of 'with' block.
        exc_type/val/tb are None if no exception occurred.
        """
        if exc_type is None:
            self.session.commit()    # success — commit
        else:
            self.session.rollback()  # failure — rollback
        self.session.close()
        return False   # do not suppress exceptions

# Usage
with DatabaseSession(engine) as db:
    post = Post(title="Hello", author_id=1)
    db.add(post)
# commit on success, rollback + close on any exception ✓

@contextmanager — Function-Based Context Managers #

from contextlib import contextmanager
import time, logging

logger = logging.getLogger(__name__)

@contextmanager
def timer(label: str):
    """Measure execution time of the block."""
    start = time.perf_counter()
    try:
        yield   # code inside 'with' block runs here
    finally:
        elapsed = time.perf_counter() - start
        logger.info(f"{label} took {elapsed:.3f}s")

with timer("database query"):
    results = db.query(Post).all()
# logs: "database query took 0.012s"

# FastAPI database session — the standard pattern
@contextmanager
def get_db_session(engine):
    session = Session(engine)
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()   # ALWAYS closed ✓

# FastAPI Depends() version — same pattern as a generator
from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db           # FastAPI provides this to route handlers
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()         # always closed ✓

# Route handler
@app.get("/posts")
async def get_posts(db: Session = Depends(get_db)):
    return db.query(Post).all()
# db session opened before handler, closed after — guaranteed ✓

contextlib Utilities #

from contextlib import suppress, nullcontext, asynccontextmanager

# suppress — silently ignore specific exceptions
with suppress(FileNotFoundError):
    Path("optional_file.txt").unlink()   # delete if exists, ignore if not

# nullcontext — a no-op context manager (useful for conditional with)
def process(data, lock=None):
    ctx = lock if lock is not None else nullcontext()
    with ctx:
        return transform(data)

# asynccontextmanager — for async context managers (FastAPI async routes)
@asynccontextmanager
async def lifespan(app):
    """FastAPI lifespan — runs on startup and shutdown."""
    # Startup
    print("Application starting up...")
    await db_pool.connect()
    yield   # app runs here
    # Shutdown
    print("Application shutting down...")
    await db_pool.disconnect()

app = FastAPI(lifespan=lifespan)

Common Mistakes #

Mistake 1 — Forgetting yield in @contextmanager #

❌ Wrong — no yield means no “inside the with block” phase:

@contextmanager
def managed():
    setup()
    # forgot yield! — raises RuntimeError: generator didn't yield

✅ Correct — always include yield:

@contextmanager
def managed():
    setup()
    yield            # ✓ execution enters the with block here
    teardown()

Mistake 2 — Using the resource after the with block #

❌ Wrong — accessing closed file:

with open("data.txt") as f:
    pass   # nothing useful done inside
content = f.read()   # ValueError: I/O operation on closed file!

✅ Correct — use the resource inside the with block:

with open("data.txt", encoding="utf-8") as f:
    content = f.read()   # ✓ inside the block

Mistake 3 — Not using finally in @contextmanager teardown #

❌ Wrong — teardown skipped when yield raises:

@contextmanager
def session():
    db = get_session()
    yield db
    db.close()   # NOT called if yield block raises an exception!

✅ Correct — wrap yield in try/finally:

@contextmanager
def session():
    db = get_session()
    try:
        yield db
    finally:
        db.close()   # ✓ ALWAYS called

Quick Reference #