Advanced Context Managers

Async context managers, nested contexts, and custom implementations.

Overview

Master advanced context manager patterns.

Async Context Managers

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_resource():
    print("Acquiring resource")
    await asyncio.sleep(0.1)
    resource = {"status": "active"}
    try:
        yield resource
    finally:
        print("Releasing resource")

async def main():
    async with async_resource() as res:
        print(f"Using: {res}")

asyncio.run(main())

Nested Context Managers

from contextlib import ExitStack

def nested_contexts():
    with ExitStack() as stack:
        files = [
            stack.enter_context(open(f, 'w'))
            for f in ['file1.txt', 'file2.txt', 'file3.txt']
        ]
        for f in files:
            f.write("Hello\n")

# Or using contextlib
from contextlib import nested

with nested(open('a.txt', 'w'), open('b.txt', 'w')) as (f1, f2):
    f1.write("Hello")
    f2.write("World")

Custom Context Managers

class DatabaseTransaction:
    def __init__(self, connection):
        self.connection = connection
        self.transaction = None
    
    def __enter__(self):
        self.transaction = self.connection.begin()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.transaction.commit()
        else:
            self.transaction.rollback()
        return False
    
    def execute(self, query, params=None):
        return self.connection.execute(query, params)

# Usage
with DatabaseTransaction(conn) as tx:
    tx.execute("INSERT INTO users VALUES (?, ?)", (1, "Alice"))
    tx.execute("INSERT INTO users VALUES (?, ?)", (2, "Bob"))
# Commits on success, rolls back on exception

Practice

Implement an async context manager for a connection pool.