Advanced Context Managers
Async context managers, nested contexts, and custom implementations.
Overview
Master advanced context manager patterns.
Async Context Managers
python
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
python
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
python
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.