Chapter 22 — Context Managers, Resource Ownership, and Transactional Cleanup

1. Opening problem

Consider:

file = open(
    "report.txt",
    "w",
    encoding="utf-8",
)

file.write(
    "Report"
)

file.close()

If write raises, close may never run.

Manual cleanup:

file = open(
    "report.txt",
    "w",
    encoding="utf-8",
)

try:
    file.write(
        "Report"
    )
finally:
    file.close()

Context manager:

with open(
    "report.txt",
    "w",
    encoding="utf-8",
) as file:
    file.write(
        "Report"
    )

The with statement makes resource ownership and cleanup part of the control-flow structure.

Context managers are not only for files. They model any temporary state with deterministic entry and exit.

---

2. The context-management protocol

A synchronous context manager implements:

__enter__()
__exit__()

Example:

class ManagedResource:
    def __enter__(
        self,
    ) -> "ManagedResource":
        self.open()
        return self

    def __exit__(
        self,
        exception_type,
        exception,
        traceback,
    ) -> bool:
        self.close()
        return False

Usage:

with ManagedResource() as resource:
    resource.use()

The as target receives the value returned by __enter__.

__exit__ receives exception information if the body fails.

---

3. Conceptual expansion of with

with manager as value:
    body()

is conceptually similar to:

manager_object = manager
value = manager_object.__enter__()

try:
    body()
except BaseException as error:
    suppress = manager_object.__exit__(
        type(error),
        error,
        error.__traceback__,
    )

    if not suppress:
        raise
else:
    manager_object.__exit__(
        None,
        None,
        None,
    )

The real language semantics are more precise, but this model explains:

• entry happens before the body;
• exit runs after success or failure;
• the context manager can suppress an exception;
• cleanup belongs to the manager.

---

4. Exception suppression

def __exit__(
    self,
    exception_type,
    exception,
    traceback,
) -> bool:
    return True

Returning a truthy value suppresses the exception.

This is dangerous unless the manager completely handles the failure.

Example of intentional suppression:

from contextlib import suppress


with suppress(
    FileNotFoundError
):
    path.unlink()

Use suppression only when the missing file is truly acceptable.

Do not suppress broad Exception.

---

5. Transaction context manager

class Transaction:
    def __init__(
        self,
        connection,
    ) -> None:
        self._connection = (
            connection
        )

    def __enter__(
        self,
    ) -> "Transaction":
        self._connection.begin()
        return self

    def __exit__(
        self,
        exception_type,
        exception,
        traceback,
    ) -> bool:
        if exception is None:
            self._connection.commit()
        else:
            self._connection.rollback()

        return False

Usage:

with Transaction(
    connection
):
    repository.save(order)
    outbox.add(event)

Questions:

• What happens if commit fails?
• What happens if rollback fails?
• Is the connection returned to a pool?
• Can the transaction be nested?
• Is the manager reusable?
• Does cancellation trigger rollback?

Context managers make structure visible, but semantics still require careful design.

---

6. Cleanup failure

Suppose the body raises and cleanup also raises.

def __exit__(...):
    self.close()

If close() raises, it can replace or chain with the original failure.

A robust implementation may preserve both meanings:

def __exit__(
    self,
    exception_type,
    exception,
    traceback,
) -> bool:
    try:
        self.close()
    except Exception as cleanup_error:
        if exception is not None:
            cleanup_error.add_note(
                "Cleanup failed while "
                "another exception was active"
            )
        raise

    return False

There is no universal rule. The cleanup failure may be more operationally urgent, but the original failure must remain diagnosable.

---

7. contextmanager

A generator can define a context manager:

from contextlib import contextmanager


@contextmanager
def temporary_directory(
    path: Path,
):
    path.mkdir(
        parents=True,
        exist_ok=False,
    )

    try:
        yield path
    finally:
        shutil.rmtree(path)

Code before yield is entry.

The yielded value becomes the as value.

Code after yield is exit and cleanup.

There must be exactly one normal yield.

---

8. Exception handling inside generator context managers

@contextmanager
def translated_errors():
    try:
        yield
    except DriverError as error:
        raise RepositoryError(
            "Repository operation failed"
        ) from error

The body exception is thrown back into the generator at yield.

Important:

If the generator catches an exception and does not re-raise, it suppresses it.

Bad:

@contextmanager
def unsafe():
    try:
        yield
    except Exception:
        logger.exception(
            "Failed"
        )

The caller may believe the operation succeeded.

---

9. Class or generator?

Use a class-based manager when:

• state is substantial;
• several methods are exposed;
• entry and exit need complex branching;
• the object may support reuse;
• typing a class clarifies the API.

Use @contextmanager when:

• one resource is acquired and released;
• the flow is linear;
• the implementation is small;
• no special manager object behavior is needed.

Do not force everything into generator syntax.

---

10. Reusable and reentrant managers

A single-use manager supports one entry.

A reusable manager can be used in separate with statements.

A reentrant manager can be entered again before a previous exit.

These are different guarantees.

Example lock:

lock = threading.RLock()

with lock:
    with lock:
        ...

An RLock is reentrant for the owning thread.

A normal file object is not designed as a generally reentrant context manager.

Document the lifecycle.

---

11. closing

Some objects expose close() but not the context-manager protocol.

from contextlib import closing


with closing(
    create_legacy_resource()
) as resource:
    resource.use()

Prefer native context-manager support when you control the class.

closing adapts an existing closeable object.

---

12. nullcontext

Sometimes a context manager is optional.

from contextlib import nullcontext


manager = (
    open(path)
    if path is not None
    else nullcontext(
        default_stream
    )
)

with manager as stream:
    process(stream)

nullcontext performs no special cleanup and yields the supplied value.

It is useful for conditional ownership without duplicating the body.

---

13. Redirecting global state

contextlib includes tools such as output redirection.

from contextlib import (
    redirect_stdout,
)

These can modify global process state.

Global redirection can be unsuitable in:

• library code;
• threaded programs;
• concurrent tests;
• long-lived services.

A context manager does not automatically make a global side effect isolated or thread-safe.

Prefer dependency injection of output streams where possible.

---

14. ExitStack

A fixed number of managers:

with open(path_a) as first, \
     open(path_b) as second:
    ...

A dynamic number:

from contextlib import ExitStack


with ExitStack() as stack:
    files = [
        stack.enter_context(
            open(
                path,
                encoding="utf-8",
            )
        )
        for path in paths
    ]

    process(files)

ExitStack records cleanup callbacks and unwinds them in reverse order.

Use cases:

• dynamic resources;
• conditional resources;
• combining callbacks and context managers;
• partially successful acquisition;
• transferring cleanup ownership.

---

15. Callback cleanup

with ExitStack() as stack:
    resource = acquire()

    stack.callback(
        resource.release
    )

    use(resource)

Callbacks are run during stack exit.

Unlike full context-manager exit methods, ordinary callbacks do not receive exception details and cannot suppress exceptions.

This often makes cleanup behavior easier to reason about.

---

16. Acquiring multiple resources safely

with ExitStack() as stack:
    connections = []

    for database in databases:
        connection = (
            stack.enter_context(
                database.connect()
            )
        )
        connections.append(
            connection
        )

    run_migration(
        connections
    )

If the third acquisition fails, the first two are cleaned up automatically.

Manual acquisition loops frequently leak earlier resources after partial failure.

---

17. Transferring ownership

ExitStack.pop_all() transfers registered cleanup callbacks to another stack.

This is an advanced ownership technique.

Use it when a function acquires resources but deliberately returns an object responsible for later cleanup.

Document ownership clearly. Hidden lifetime transfer causes leaks.

---

18. Async context managers

An asynchronous context manager implements:

__aenter__()
__aexit__()

Usage:

async with client.session() as session:
    await session.send(
        request
    )

Example:

class AsyncConnection:
    async def __aenter__(
        self,
    ) -> "AsyncConnection":
        await self.connect()
        return self

    async def __aexit__(
        self,
        exception_type,
        exception,
        traceback,
    ) -> bool:
        await self.close()
        return False

Asynchronous cleanup can await network or asynchronous synchronization operations.

---

19. asynccontextmanager

from contextlib import (
    asynccontextmanager,
)


@asynccontextmanager
async def managed_client():
    client = Client()
    await client.open()

    try:
        yield client
    finally:
        await client.close()

Usage:

async with managed_client() as client:
    await client.fetch()

The same exception-suppression warning applies: catching and not re-raising suppresses body failures.

---

20. AsyncExitStack

from contextlib import (
    AsyncExitStack,
)


async with AsyncExitStack() as stack:
    clients = []

    for endpoint in endpoints:
        client = await (
            stack.enter_async_context(
                connect(endpoint)
            )
        )
        clients.append(client)

    await use_all(clients)

It supports dynamic mixtures of asynchronous cleanup.

Use it for:

• multiple async sessions;
• dynamic subscriptions;
• temporary servers;
• async locks;
• mixed sync and async callback ownership.

---

21. Cancellation-safe cleanup

Async tasks can be cancelled at an await.

async def worker() -> None:
    resource = await acquire()

    try:
        await use(resource)
    finally:
        await release(resource)

Cleanup itself includes an await and can also be affected by cancellation.

The correct design depends on the library and resource semantics.

Questions:

• must cleanup complete despite cancellation?
• is the close operation idempotent?
• should cleanup use a timeout?
• should cancellation be temporarily shielded?
• can the resource be abandoned safely?
• who owns final cleanup?

Avoid casually swallowing CancelledError.

Cancellation is control flow, not merely an ordinary error.

---

22. Context variables

Temporary contextual state in asynchronous programs should not be stored in ordinary globals or assumed to be thread-local.

from contextvars import ContextVar


request_id = ContextVar[
    str
](
    "request_id"
)

Temporary binding:

token = request_id.set(
    "req-123"
)

try:
    process()
finally:
    request_id.reset(
        token
    )

A context manager can wrap this pattern.

Context variables propagate according to execution context and work better with asynchronous tasks than plain thread-local state.

---

23. Dependency injection versus global context

A request ID may fit context-local logging.

A database session usually should remain an explicit dependency.

Do not use context managers and context variables to hide every dependency.

Ask:

• Is this ambient execution metadata?
• Does every function genuinely need it?
• Will tests understand ownership?
• Does hidden state cross task boundaries correctly?
• Is explicit parameter passing clearer?

---

24. Context managers as decorators

ContextDecorator can allow a context manager to wrap a function.

from contextlib import (
    ContextDecorator,
)


class traced(
    ContextDecorator
):
    def __enter__(self):
        start_trace()
        return self

    def __exit__(
        self,
        *details,
    ):
        end_trace()
        return False

Usage:

@traced()
def process() -> None:
    ...

The manager must support the required reuse semantics because decorated functions may be called repeatedly.

Explicit with is often clearer when scope matters.

---

25. TypeScript comparison

TypeScript traditionally uses:

const resource =
  await acquire();

try {
  await use(resource);
} finally {
  await resource.close();
}

Modern JavaScript environments also have evolving resource-management features, but support and target environments vary.

Python's with and async with are mature language protocols.

The important design issue in both languages is ownership:

• who acquires;
• who releases;
• when cleanup runs;
• whether cleanup can fail;
• whether cancellation interrupts cleanup.

---

26. Common mistakes

Assuming context manager means thread-safe

It only structures entry and exit.

Returning True accidentally

This suppresses exceptions.

Catching without re-raising in @contextmanager

Also suppresses.

Returning a lazy object tied to an exited manager

The resource is already closed.

Fixed nested with for dynamic resources

Use ExitStack.

Global-state redirection in concurrent code

Prefer explicit dependencies.

Hidden resource ownership

Document who closes returned objects.

Ignoring cancellation during async cleanup

Design cleanup semantics deliberately.

---

27. English vocabulary

Term	Meaning
context manager	object controlling entry and exit of a scope
deterministic cleanup	cleanup tied to explicit control flow
ownership	responsibility for releasing a resource
suppression	preventing an exception from propagating
reentrant	safe to enter again before leaving
reusable	safe for repeated separate uses
unwinding	running cleanup in reverse order
partial acquisition	failure after some resources were obtained
ambient context	state available without explicit parameters
cancellation-safe	preserving invariants when async work is cancelled

Useful sentences:

• “The context manager owns the resource for the duration of the block.”
• “Returning a truthy value from __exit__ suppresses the exception.”
• “The stack unwinds successfully acquired resources in reverse order.”
• “The generator manager must re-raise failures from the managed body.”
• “Global output redirection is unsafe in concurrent library code.”
• “Asynchronous cleanup must define its cancellation semantics.”

---

28. Speaking task

Explain for ten minutes:

Why context managers are fundamentally about ownership, not syntax convenience.

Include synchronous, asynchronous, dynamic, and transactional examples.

---

29. Writing task

Write a 450-word design review of an asynchronous service that opens database sessions in helper functions and expects garbage collection to close them later.

---

30. Exercises

Exercise 1

Implement a class-based timer context manager.

Exercise 2

Implement a generator-based temporary environment-variable manager.

Exercise 3

Use ExitStack to open a dynamic collection of files.

Exercise 4

Implement an async client context manager.

Exercise 5

Design a transaction manager that preserves body failures when rollback also fails.

Exercise 6

Wrap a ContextVar binding in a context manager.

---

31. Complete solutions

Solution 1

from time import perf_counter


class Timer:
    def __enter__(
        self,
    ) -> "Timer":
        self._started = (
            perf_counter()
        )
        self.elapsed = 0.0
        return self

    def __exit__(
        self,
        exception_type,
        exception,
        traceback,
    ) -> bool:
        self.elapsed = (
            perf_counter()
            - self._started
        )
        return False

Solution 2

import os
from contextlib import (
    contextmanager,
)


_MISSING = object()


@contextmanager
def temporary_environment(
    name: str,
    value: str,
):
    previous = os.environ.get(
        name,
        _MISSING,
    )

    os.environ[name] = value

    try:
        yield
    finally:
        if previous is _MISSING:
            os.environ.pop(
                name,
                None,
            )
        else:
            os.environ[name] = (
                previous
            )

This changes process-global state and is not isolated across threads or concurrent tests.

Solution 3

from contextlib import ExitStack


def read_all(
    paths: list[Path],
) -> list[str]:
    with ExitStack() as stack:
        files = [
            stack.enter_context(
                path.open(
                    encoding="utf-8"
                )
            )
            for path in paths
        ]

        return [
            file.read()
            for file in files
        ]

Solution 4

from contextlib import (
    asynccontextmanager,
)


@asynccontextmanager
async def managed_client():
    client = ApiClient()
    await client.connect()

    try:
        yield client
    finally:
        await client.close()

Solution 5

class Transaction:
    def __init__(
        self,
        connection,
    ) -> None:
        self._connection = (
            connection
        )

    def __enter__(
        self,
    ) -> "Transaction":
        self._connection.begin()
        return self

    def __exit__(
        self,
        exception_type,
        exception,
        traceback,
    ) -> bool:
        if exception is None:
            self._connection.commit()
            return False

        try:
            self._connection.rollback()
        except Exception as rollback_error:
            rollback_error.add_note(
                "Rollback failed after "
                f"{type(exception).__name__}"
            )
            raise rollback_error from exception

        return False

The design preserves a causal chain. Some systems may prefer an exception group when both failures need equal visibility.

Solution 6

from contextlib import (
    contextmanager,
)
from contextvars import ContextVar


request_id = ContextVar[
    str
](
    "request_id"
)


@contextmanager
def request_context(
    value: str,
):
    token = request_id.set(
        value
    )

    try:
        yield
    finally:
        request_id.reset(
            token
        )

---

32. Chapter checkpoint

You should now be able to explain:

1. context-manager protocol;
2. exception suppression;
3. transaction managers;
4. generator context managers;
5. cleanup failures;
6. reusable and reentrant managers;
7. nullcontext and closing;
8. ExitStack;
9. async context managers;
10. AsyncExitStack;
11. cancellation-safe cleanup;
12. context variables;
13. ownership transfer.