Exceptions
Context Managers
Automatic Cleanup
You open a file, read it, then must close it. Forgetting to close leaks resources.
Context managers with with automatically close resources when done - even if
exceptions occur. No more manual try-finally for cleanup.
Basic with statement
Let Python handle resource cleanup.
# Basic with statement usage
def main():
# Simulated file handling
class FakeFile:
def __init__(self, name, mode):
self.name = name
self.mode = mode
self.closed = False
def __enter__(self):
print(f" Opening {self.name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" Closing {self.name}")
self.closed = True
return False # Don't suppress exceptions
def read(self):
if self.closed:
raise ValueError("File is closed")
return f"Contents of {self.name}"
print("Basic with statement:\n")
# Using with statement
with FakeFile("data.txt", "r") as f:
content = f.read()
print(f" Read: {content}")
print(f" File closed: {f.closed}")
# Without with statement (manual)
print("\nManual resource management:")
f2 = FakeFile("manual.txt", "r")
f2.__enter__()
try:
content = f2.read()
print(f" Read: {content}")
finally:
f2.__exit__(None, None, None)
# Multiple context managers
print("\nMultiple resources:")
with FakeFile("file1.txt", "r") as f1, FakeFile("file2.txt", "r") as f2:
print(f" Reading {f1.name}")
print(f" Reading {f2.name}")
print(" Both files closed")
# Exception handling
print("\nWith exception:")
try:
with FakeFile("error.txt", "r") as f:
print(f" Inside with block")
raise ValueError("Simulated error")
except ValueError as e:
print(f" Caught: {e}")
print(f" File was still closed: {f.closed}")
# Timer context
class Timer:
def __enter__(self):
self.start = 1000.0
print(" Timer started")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.elapsed = 1000.0001 - self.start
print(f" Timer stopped: {self.elapsed:.4f}s")
return False
print("\nTiming code:")
with Timer():
# Simulate work
total = sum(range(1000))
print(f" Computed sum: {total}")
if __name__ == "__main__":
main()
def main(): # Simulated file handling
3def main():4 # Simulated file handling5 class FakeFile:6 def __init__(self, name, mode):7 self.name = name8 self.mode = mode9 self.closed = False10 11 def __enter__(self):12 print(f" Opening {self.name}")13 return self14 15 def __exit__(self, exc_type, exc_val, exc_tb):16 print(f" Closing {self.name}")17 self.closed = True18 return False # Don't suppress exceptions19 20 def read(self):21 if self.closed:22 raise ValueError("File is closed")23 return f"Contents of {self.name}"24 25 print("Basic with statement:\n")outputBasic with statement:self.name ← data.txt, self.mode ← r, self.closed ← False
pass 1 of 55class FakeFile:6 def __init__(self<__main__.main.<locals>.FakeFile object at ⟨addr A⟩>, namedata.txt, moder):7 self.name→ data.txt = namedata.txt8 self.mode→ r = moder9 self.closed→ False = FalseAll 5 passes — pass 1 is the card above pass selfnameff1.namef2.nameeself.nameself.modeself.closed1 <__main__.main.<locals>.FakeFile object at ⟨addr A⟩> data.txt <__main__.main.<locals>.FakeFile object at ⟨addr A⟩> — — — data.txt r False 2 <__main__.main.<locals>.FakeFile object at ⟨addr B⟩> manual.txt — — — — manual.txt r False 3 <__main__.main.<locals>.FakeFile object at ⟨addr C⟩> file1.txt — — — — file1.txt r False 4 <__main__.main.<locals>.FakeFile object at ⟨addr D⟩> file2.txt — file1.txt file2.txt — file2.txt r False 5 <__main__.main.<locals>.FakeFile object at ⟨addr B⟩> error.txt — — — Simulated error error.txt r False def __enter__(self):
pass 1 of 511def __enter__(self<__main__.main.<locals>.FakeFile object at ⟨addr A⟩>):12 print(f" Opening {self.namedata.txt}")13 return selfoutput Opening data.txtAll 5 passes — pass 1 is the card above pass selfself.nameff1.namef2.namee1 <__main__.main.<locals>.FakeFile object at ⟨addr A⟩> data.txt <__main__.main.<locals>.FakeFile object at ⟨addr A⟩> — — — 2 <__main__.main.<locals>.FakeFile object at ⟨addr B⟩> manual.txt — — — — 3 <__main__.main.<locals>.FakeFile object at ⟨addr C⟩> file1.txt — — — — 4 <__main__.main.<locals>.FakeFile object at ⟨addr D⟩> file2.txt — file1.txt file2.txt — 5 <__main__.main.<locals>.FakeFile object at ⟨addr B⟩> error.txt — — — Simulated error with FakeFile("data.txt", "r") as f:
27# Using with statement28with FakeFile("data.txt", "r") as f:29 content = f<__main__.main.<locals>.FakeFile object at ⟨addr A⟩>.read()30 print(f" Read: {content}")def read(self):
pass 1 of 220def read(self<__main__.main.<locals>.FakeFile object at ⟨addr A⟩>):21 if self.closed:22 raise ValueError("File is closed")23 return f"Contents of {self.namedata.txt}"content ← Contents of data.txt
28with FakeFile("data.txt", "r") as f:29 content→ Contents of data.txt = f<__main__.main.<locals>.FakeFile object at ⟨addr A⟩>.read()30 print(f" Read: {contentContents of data.txt}")output Read: Contents of data.txtself.closed ← True
pass 1 of 515def __exit__(self<__main__.main.<locals>.FakeFile object at ⟨addr A⟩>, exc_typeNone, exc_valNone, exc_tbNone):16 print(f" Closing {self.namedata.txt}")17 self.closed→ True = True18 return False # Don't suppress exceptionsoutput Closing data.txtAll 5 passes — pass 1 is the card above pass selfexc_typeexc_valexc_tbself.nameeself.closed1 <__main__.main.<locals>.FakeFile object at ⟨addr A⟩> None None None data.txt — True 2 <__main__.main.<locals>.FakeFile object at ⟨addr B⟩> None None None manual.txt — True 3 <__main__.main.<locals>.FakeFile object at ⟨addr D⟩> None None None file2.txt — True 4 <__main__.main.<locals>.FakeFile object at ⟨addr C⟩> None None None file1.txt — True 5 <__main__.main.<locals>.FakeFile object at ⟨addr B⟩> <class 'ValueError'> Simulated error ⟨traceback E⟩ error.txt Simulated error True print(f" File closed: {f.closed}")
32print(f" File closed: {f.closedTrue}")3334#@help h135# with statement calls __enter__ on entry36# Calls __exit__ on exit (even if exception occurs)37# Guarantees cleanup happens38#@end3940# Without with statement (manual)41print("\nManual resource management:")4243f2 = FakeFile("manual.txt", "r")44f2.__enter__()output File closed: True Manual resource management:f2 ← <__main__.main.<locals>.FakeFile object at ⟨addr B⟩>
43f2→ <__main__.main.<locals>.FakeFile object at ⟨addr B⟩> = FakeFile("manual.txt", "r")44f2<__main__.main.<locals>.FakeFile object at ⟨addr B⟩>.__enter__()45try:f2.__enter__()
43f2 = FakeFile("manual.txt", "r")44f2<__main__.main.<locals>.FakeFile object at ⟨addr B⟩>.__enter__()45try:try:
44f2.__enter__()45try:46 content = f2<__main__.main.<locals>.FakeFile object at ⟨addr B⟩>.read()47 print(f" Read: {content}")def read(self):
pass 2 of 220def read(self<__main__.main.<locals>.FakeFile object at ⟨addr B⟩>):21 if self.closed:22 raise ValueError("File is closed")23 return f"Contents of {self.namemanual.txt}"content ← Contents of manual.txt
45try:46 content→ Contents of manual.txt = f2<__main__.main.<locals>.FakeFile object at ⟨addr B⟩>.read()47 print(f" Read: {contentContents of manual.txt}")48finally:output Read: Contents of manual.txtfinally:
46 content = f2.read()47 print(f" Read: {content}")48finally:49 f2<__main__.main.<locals>.FakeFile object at ⟨addr B⟩>.__exit__(None, None, None)f2.__exit__(None, None, None)
48finally:49 f2<__main__.main.<locals>.FakeFile object at ⟨addr B⟩>.__exit__(None, None, None)print(" Multiple resources:")
51# Multiple context managers52print("\nMultiple resources:")output Multiple resources:with FakeFile("file1.txt", "r") as f1, FakeFile("file2.txt", "r") as f…
54with FakeFile("file1.txt", "r") as f1, FakeFile("file2.txt", "r") as f2:55 print(f" Reading {f1.namefile1.txt}")56 print(f" Reading {f2.namefile2.txt}")output Reading file1.txt Reading file2.txtprint(" Both files closed")
58print(" Both files closed")5960# Exception handling61print("\nWith exception:")output Both files closed With exception:with FakeFile("error.txt", "r") as f:
63try:64 with FakeFile("error.txt", "r") as f:65 print(f" Inside with block")66 raise ValueError("Simulated error")67except ValueError as e:output Inside with blockexcept ValueError as e:
66 raise ValueError("Simulated error")67except ValueError as e:68 print(f" Caught: {eSimulated error}")6970print(f" File was still closed: {f.closed}")output Caught: Simulated error Caught: Simulated errorprint(f" File was still closed: {f.closed}")
70print(f" File was still closed: {f.closedTrue}")7172# Timer context73class Timer:74 def __enter__(self):75 self.start = 1000.076 print(" Timer started")77 return self78 79 def __exit__(self, exc_type, exc_val, exc_tb):80 self.elapsed = 1000.0001 - self.start81 print(f" Timer stopped: {self.elapsed:.4f}s")82 return False8384print("\nTiming code:")output File was still closed: True Timing code:self.start ← 1000.0
73class Timer:74 def __enter__(self<__main__.main.<locals>.Timer object at ⟨addr A⟩>):75 self.start→ 1000.0 = 1000.076 print(" Timer started")77 return selfoutput Timer startedtotal ← 499500
86with Timer():87 # Simulate work88 total→ 499500 = sum(range(1000))89 print(f" Computed sum: {total499500}")output Computed sum: 499500self.elapsed ← 9.999999997489795e-05
79def __exit__(self<__main__.main.<locals>.Timer object at ⟨addr A⟩>, exc_typeNone, exc_valNone, exc_tbNone):80 self.elapsed→ 9.999999997489795e-05 = 1000.0001 - self.start1000.081 print(f" Timer stopped: {self.elapsed9.999999997489795e-05:.4f}s")82 return Falseoutput Timer stopped: 0.0001smain()
91if __name__ == "__main__":92 main()
with open(file) as f: automatically closes file when block exits.
Class-based context manager
Create your own with enter and exit.
# Class-based context managers
def main():
# Simple resource manager
class Resource:
def __init__(self, name):
self.name = name
self.acquired = False
def __enter__(self):
print(f" Acquiring {self.name}")
self.acquired = True
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" Releasing {self.name}")
self.acquired = False
return False # Propagate exceptions
def use(self):
if not self.acquired:
raise RuntimeError("Resource not acquired")
print(f" Using {self.name}")
print("Resource management:\n")
with Resource("Database") as db:
db.use()
print(f" Acquired: {db.acquired}")
print(f" After with: {db.acquired}")
# Context manager with return value
class Connection:
def __init__(self, host, port):
self.host = host
self.port = port
self.connected = False
def __enter__(self):
print(f" Connecting to {self.host}:{self.port}")
self.connected = True
return self # Returns self for use in 'as' clause
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" Disconnecting from {self.host}:{self.port}")
self.connected = False
return False
def send(self, data):
if not self.connected:
raise RuntimeError("Not connected")
print(f" Sent: {data}")
print("\nConnection:")
with Connection("localhost", 8080) as conn:
conn.send("Hello")
conn.send("World")
# Exception information in __exit__
class ErrorHandler:
def __enter__(self):
print(" Entering context")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
print(" Exiting normally")
else:
print(f" Exception occurred: {exc_type.__name__}")
print(f" Message: {exc_val}")
return False # Don't suppress
print("\nError handling:")
with ErrorHandler():
print(" No error")
print()
try:
with ErrorHandler():
print(" About to raise error")
raise ValueError("Test error")
except ValueError:
print(" Exception propagated")
# Suppress exception by returning True
class Suppressor:
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is ValueError:
print(f" Suppressing {exc_type.__name__}: {exc_val}")
return True # Suppress ValueError
return False # Propagate other exceptions
print("\nSuppressing exceptions:")
with Suppressor():
print(" Raising ValueError")
raise ValueError("This will be suppressed")
print(" Execution continued")
# Lock-like context manager
class Lock:
def __init__(self, name):
self.name = name
self.locked = False
def __enter__(self):
print(f" Acquiring lock: {self.name}")
self.locked = True
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" Releasing lock: {self.name}")
self.locked = False
return False
print("\nLocking:")
lock = Lock("data_lock")
with lock:
print(f" Lock held: {lock.locked}")
# Critical section
print(f" Lock released: {lock.locked}")
if __name__ == "__main__":
main()
def main(): # Simple resource manager
3def main():4 # Simple resource manager5 class Resource:6 def __init__(self, name):7 self.name = name8 self.acquired = False9 10 def __enter__(self):11 print(f" Acquiring {self.name}")12 self.acquired = True13 return self14 15 def __exit__(self, exc_type, exc_val, exc_tb):16 print(f" Releasing {self.name}")17 self.acquired = False18 return False # Propagate exceptions19 20 def use(self):21 if not self.acquired:22 raise RuntimeError("Resource not acquired")23 print(f" Using {self.name}")24 25 print("Resource management:\n")outputResource management:self.name ← Database, self.acquired ← False
5class Resource:6 def __init__(self<__main__.main.<locals>.Resource object at ⟨addr A⟩>, nameDatabase):7 self.name→ Database = nameDatabase8 self.acquired→ False = Falseself.acquired ← True
10def __enter__(self<__main__.main.<locals>.Resource object at ⟨addr A⟩>):11 print(f" Acquiring {self.nameDatabase}")12 self.acquired→ True = True13 return selfoutput Acquiring Databasewith Resource("Database") as db:
27with Resource("Database") as db:28 db<__main__.main.<locals>.Resource object at ⟨addr A⟩>.use()29 print(f" Acquired: {db.acquired}")def use(self):
20def use(self<__main__.main.<locals>.Resource object at ⟨addr A⟩>):21 if not self.acquired:22 raise RuntimeError("Resource not acquired")23 print(f" Using {self.nameDatabase}")output Using Databasedb.use()
27with Resource("Database") as db:28 db<__main__.main.<locals>.Resource object at ⟨addr A⟩>.use()29 print(f" Acquired: {db.acquiredTrue}")output Acquired: Trueself.acquired ← False
15def __exit__(self<__main__.main.<locals>.Resource object at ⟨addr A⟩>, exc_typeNone, exc_valNone, exc_tbNone):16 print(f" Releasing {self.nameDatabase}")17 self.acquired→ False = False18 return False # Propagate exceptionsoutput Releasing Databaseprint(f" After with: {db.acquired}")
31print(f" After with: {db.acquiredFalse}")3233#@help h134# __enter__() called at start of with block35# __exit__(exc_type, exc_val, exc_tb) called at end36# exc_* parameters are None if no exception37# Return True from __exit__ to suppress exception38#@end3940# Context manager with return value41class Connection:42 def __init__(self, host, port):43 self.host = host44 self.port = port45 self.connected = False46 47 def __enter__(self):48 print(f" Connecting to {self.host}:{self.port}")49 self.connected = True50 return self # Returns self for use in 'as' clause51 52 def __exit__(self, exc_type, exc_val, exc_tb):53 print(f" Disconnecting from {self.host}:{self.port}")54 self.connected = False55 return False56 57 def send(self, data):58 if not self.connected:59 raise RuntimeError("Not connected")60 print(f" Sent: {data}")6162print("\nConnection:")output After with: False Connection:self.host ← localhost, self.port ← 8080, self.connected ← False
41class Connection:42 def __init__(self<__main__.main.<locals>.Connection object at ⟨addr B⟩>, hostlocalhost, port8080):43 self.host→ localhost = hostlocalhost44 self.port→ 8080 = port808045 self.connected→ False = Falseself.connected ← True
47def __enter__(self<__main__.main.<locals>.Connection object at ⟨addr B⟩>):48 print(f" Connecting to {self.hostlocalhost}:{self.port8080}")49 self.connected→ True = True50 return self # Returns self for use in 'as' clauseoutput Connecting to localhost:8080with Connection("localhost", 8080) as conn:
64with Connection("localhost", 8080) as conn:65 conn<__main__.main.<locals>.Connection object at ⟨addr B⟩>.send("Hello")66 conn.send("World")def send(self, data):
pass 1 of 257def send(self<__main__.main.<locals>.Connection object at ⟨addr B⟩>, dataHello):58 if not self.connected:59 raise RuntimeError("Not connected")60 print(f" Sent: {dataHello}")output Sent: Helloconn.send("Hello")
64with Connection("localhost", 8080) as conn:65 conn<__main__.main.<locals>.Connection object at ⟨addr B⟩>.send("Hello")66 conn<__main__.main.<locals>.Connection object at ⟨addr B⟩>.send("World")def send(self, data):
pass 2 of 257def send(self<__main__.main.<locals>.Connection object at ⟨addr B⟩>, dataWorld):58 if not self.connected:59 raise RuntimeError("Not connected")60 print(f" Sent: {dataWorld}")output Sent: Worldconn.send("World")
65conn.send("Hello")66conn<__main__.main.<locals>.Connection object at ⟨addr B⟩>.send("World")self.connected ← False
52def __exit__(self<__main__.main.<locals>.Connection object at ⟨addr B⟩>, exc_typeNone, exc_valNone, exc_tbNone):53 print(f" Disconnecting from {self.hostlocalhost}:{self.port8080}")54 self.connected→ False = False55 return Falseoutput Disconnecting from localhost:8080# Exception information in __exit__
68# Exception information in __exit__69class ErrorHandler:70 def __enter__(self):71 print(" Entering context")72 return self73 74 def __exit__(self, exc_type, exc_val, exc_tb):75 if exc_type is None:76 print(" Exiting normally")77 else:78 print(f" Exception occurred: {exc_type.__name__}")79 print(f" Message: {exc_val}")80 return False # Don't suppress8182print("\nError handling:")output Error handling:def __enter__(self):
pass 1 of 269class ErrorHandler:70 def __enter__(self<__main__.main.<locals>.ErrorHandler object at ⟨addr C⟩>):71 print(" Entering context")72 return selfoutput Entering contextwith ErrorHandler():
84with ErrorHandler():85 print(" No error")output No errordef __exit__(self, exc_type, exc_val, exc_tb):
pass 1 of 274def __exit__(self<__main__.main.<locals>.ErrorHandler object at ⟨addr C⟩>, exc_typeNone, exc_valNone, exc_tbNone):75 if exc_type is None:76 print(" Exiting normally")if exc_type is None:
74def __exit__(self, exc_type, exc_val, exc_tb):75 if exc_typeNone is None:76 print(" Exiting normally")77 else:output Exiting normallyreturn False # Don't suppress
79 print(f" Message: {exc_val}")80return False # Don't suppressprint()
87print()def __enter__(self):
pass 2 of 269class ErrorHandler:70 def __enter__(self<__main__.main.<locals>.ErrorHandler object at ⟨addr C⟩>):71 print(" Entering context")72 return selfoutput Entering contextwith ErrorHandler():
89try:90 with ErrorHandler():91 print(" About to raise error")92 raise ValueError("Test error")93except ValueError:output About to raise errordef __exit__(self, exc_type, exc_val, exc_tb):
pass 2 of 274def __exit__(self<__main__.main.<locals>.ErrorHandler object at ⟨addr C⟩>, exc_type<class 'ValueError'>, exc_valTest error, exc_tb⟨traceback D⟩):75 if exc_type is None:76 print(" Exiting normally")else:
75if exc_type is None:76 print(" Exiting normally")77else:78 print(f" Exception occurred: {exc_type.__name__ValueError}")79 print(f" Message: {exc_valTest error}")80return False # Don't suppressoutput Exception occurred: ValueError Message: Test errorreturn False # Don't suppress
79 print(f" Message: {exc_val}")80return False # Don't suppressexcept ValueError:
92 raise ValueError("Test error")93except ValueError:94 print(" Exception propagated")9596# Suppress exception by returning Trueoutput Exception propagated Exception propagated# Suppress exception by returning True
96# Suppress exception by returning True97class Suppressor:98 def __enter__(self):99 return self100 101 def __exit__(self, exc_type, exc_val, exc_tb):102 if exc_type is ValueError:103 print(f" Suppressing {exc_type.__name__}: {exc_val}")104 return True # Suppress ValueError105 return False # Propagate other exceptions106107print("\nSuppressing exceptions:")output Suppressing exceptions:def __enter__(self):
97class Suppressor:98 def __enter__(self<__main__.main.<locals>.Suppressor object at ⟨addr C⟩>):99 return selfwith Suppressor():
109with Suppressor():110 print(" Raising ValueError")111 raise ValueError("This will be suppressed")output Raising ValueErrordef __exit__(self, exc_type, exc_val, exc_tb):
101def __exit__(self<__main__.main.<locals>.Suppressor object at ⟨addr C⟩>, exc_type<class 'ValueError'>, exc_valThis will be suppressed, exc_tb⟨traceback E⟩):102 if exc_type is ValueError:103 print(f" Suppressing {exc_type.__name__}: {exc_val}")if exc_type is ValueError:
101def __exit__(self, exc_type, exc_val, exc_tb):102 if exc_type<class 'ValueError'> is ValueError<class 'ValueError'>:103 print(f" Suppressing {exc_type.__name__ValueError}: {exc_valThis will be suppressed}")104 return True # Suppress ValueError105 return False # Propagate other exceptionsoutput Suppressing ValueError: This will be suppressedprint(" Execution continued")
113print(" Execution continued")114115# Lock-like context manager116class Lock:117 def __init__(self, name):118 self.name = name119 self.locked = False120 121 def __enter__(self):122 print(f" Acquiring lock: {self.name}")123 self.locked = True124 return self125 126 def __exit__(self, exc_type, exc_val, exc_tb):127 print(f" Releasing lock: {self.name}")128 self.locked = False129 return False130131print("\nLocking:")132133lock = Lock("data_lock")output Execution continued Locking:self.name ← data_lock, self.locked ← False
116class Lock:117 def __init__(self<__main__.main.<locals>.Lock object at ⟨addr C⟩>, namedata_lock):118 self.name→ data_lock = namedata_lock119 self.locked→ False = Falselock ← <__main__.main.<locals>.Lock object at ⟨addr C⟩>
133lock→ <__main__.main.<locals>.Lock object at ⟨addr C⟩> = Lock("data_lock")self.locked ← True
121def __enter__(self<__main__.main.<locals>.Lock object at ⟨addr C⟩>):122 print(f" Acquiring lock: {self.namedata_lock}")123 self.locked→ True = True124 return selfoutput Acquiring lock: data_lockwith lock:
135with lock<__main__.main.<locals>.Lock object at ⟨addr C⟩>:136 print(f" Lock held: {lock.lockedTrue}")137 # Critical sectionoutput Lock held: Trueself.locked ← False
126def __exit__(self<__main__.main.<locals>.Lock object at ⟨addr C⟩>, exc_typeNone, exc_valNone, exc_tbNone):127 print(f" Releasing lock: {self.namedata_lock}")128 self.locked→ False = False129 return Falseoutput Releasing lock: data_lockprint(f" Lock released: {lock.locked}")
139print(f" Lock released: {lock.lockedFalse}")output Lock released: Falsemain()
141if __name__ == "__main__":142 main()
__enter__ sets up, __exit__ cleans up. Works with with statement.
Decorator-based context manager
Simpler syntax with @contextmanager.
# contextmanager decorator
from contextlib import contextmanager
def main():
# Simple context manager with decorator
@contextmanager
def managed_resource(name):
print(f" Setup: {name}")
yield name # Provide value to 'as' clause
print(f" Cleanup: {name}")
print("Decorator-based context manager:\n")
with managed_resource("Resource1") as res:
print(f" Using: {res}")
# Context manager with exception handling
@contextmanager
def safe_operation(name):
print(f" Starting: {name}")
try:
yield name
except Exception as e:
print(f" Error in {name}: {e}")
raise # Re-raise
finally:
print(f" Finishing: {name}")
print("\nWith exception handling:")
with safe_operation("Operation1"):
print(" Working...")
print()
try:
with safe_operation("Operation2"):
print(" Working...")
raise ValueError("Something went wrong")
except ValueError:
print(" Exception caught in main")
# Timer context manager
@contextmanager
def timer(label):
start = 1000.0
yield
elapsed = 1000.0001 - start
print(f" {label}: {elapsed:.4f}s")
print("\nTiming:")
with timer("Computation"):
total = sum(range(100000))
print(f" Sum: {total}")
# Temporary directory simulation
@contextmanager
def temp_directory(name):
print(f" Creating temp dir: {name}")
dir_path = f"/tmp/{name}"
try:
yield dir_path
finally:
print(f" Removing temp dir: {name}")
print("\nTemporary directory:")
with temp_directory("work_dir") as path:
print(f" Using directory: {path}")
# Do work in temp directory
# Database transaction simulation
@contextmanager
def transaction(db_name):
print(f" BEGIN TRANSACTION on {db_name}")
try:
yield
print(f" COMMIT on {db_name}")
except Exception as e:
print(f" ROLLBACK on {db_name} ({e})")
raise
print("\nDatabase transactions:")
with transaction("users_db"):
print(" INSERT INTO users...")
print(" UPDATE users...")
print()
try:
with transaction("orders_db"):
print(" INSERT INTO orders...")
raise RuntimeError("Constraint violation")
except RuntimeError:
print(" Transaction rolled back")
# Changing context
@contextmanager
def working_directory(path):
original = "/current/dir"
print(f" Changing to: {path}")
try:
yield path
finally:
print(f" Restoring to: {original}")
print("\nWorking directory:")
with working_directory("/new/path") as path:
print(f" Working in: {path}")
if __name__ == "__main__":
main()
def main(): # Simple context manager with decorator
5def main():6 # Simple context manager with decorator7 @contextmanager8 def managed_resource(name):9 print(f" Setup: {name}")10 yield name # Provide value to 'as' clause11 print(f" Cleanup: {name}")12 13 print("Decorator-based context manager:\n")outputDecorator-based context manager:def managed_resource(name):
7@contextmanager8def managed_resource(nameResource1):9 print(f" Setup: {nameResource1}")10 yield nameResource1 # Provide value to 'as' clause11 print(f" Cleanup: {name}")output Setup: Resource1with managed_resource("Resource1") as res:
9 print(f" Setup: {name}")10 yield nameResource1 # Provide value to 'as' clause11 print(f" Cleanup: {nameResource1}")1213print("Decorator-based context manager:\n")1415with managed_resource("Resource1") as res:16 print(f" Using: {resResource1}")output Using: Resource1 Cleanup: Resource1print(" With exception handling:")
37print("\nWith exception handling:")output With exception handling:def safe_operation(name):
pass 1 of 226@contextmanager27def safe_operation(nameOperation1):28 print(f" Starting: {nameOperation1}")29 try:output Starting: Operation1try:
pass 1 of 228print(f" Starting: {name}")29try:30 yield nameOperation131except Exception as e:with safe_operation("Operation1"):
29 try:30 yield nameOperation131 except Exception as e:32 print(f" Error in {name}: {e}")33 raise # Re-raise34 finally:35 print(f" Finishing: {name}")3637print("\nWith exception handling:")3839with safe_operation("Operation1"):40 print(" Working...")output Working...# Re-raise finally:
pass 1 of 232 print(f" Error in {name}: {e}")33 raise # Re-raise34finally:35 print(f" Finishing: {nameOperation1}")output Finishing: Operation1 Finishing: Operation1print()
42print()def safe_operation(name):
pass 2 of 226@contextmanager27def safe_operation(nameOperation2):28 print(f" Starting: {nameOperation2}")29 try:output Starting: Operation2try:
pass 2 of 228print(f" Starting: {name}")29try:30 yield nameOperation231except Exception as e:with safe_operation("Operation2"):
44try:45 with safe_operation("Operation2"):46 print(" Working...")47 raise ValueError("Something went wrong")48except ValueError:output Working...except Exception as e:
30 yield name31except Exception as e:32 print(f" Error in {nameOperation2}: {eSomething went wrong}")33 raise # Re-raise34finally:output Error in Operation2: Something went wrong# Re-raise finally:
pass 2 of 232 print(f" Error in {name}: {e}")33 raise # Re-raise34finally:35 print(f" Finishing: {nameOperation2}")output Finishing: Operation2 Finishing: Operation2except ValueError:
47 raise ValueError("Something went wrong")48except ValueError:49 print(" Exception caught in main")5051# Timer context manageroutput Exception caught in main Exception caught in mainprint(" Timing:")
59print("\nTiming:")output Timing:start ← 1000.0
52@contextmanager53def timer(labelComputation):54 start→ 1000.0 = 1000.055 yield56 elapsed = 1000.0001 - starttotal ← 4999950000, elapsed ← 9.999999997489795e-05
54 start = 1000.055 yield56 elapsed→ 9.999999997489795e-05 = 1000.0001 - start1000.057 print(f" {labelComputation}: {elapsed9.999999997489795e-05:.4f}s")5859print("\nTiming:")6061with timer("Computation"):62 total→ 4999950000 = sum(range(100000))63 print(f" Sum: {total4999950000}")output Sum: 4999950000 Computation: 0.0001sprint(" Temporary directory:")
75print("\nTemporary directory:")output Temporary directory:dir_path ← /tmp/work_dir
66@contextmanager67def temp_directory(namework_dir):68 print(f" Creating temp dir: {namework_dir}")69 dir_path→ /tmp/work_dir = f"/tmp/{namework_dir}"70 try:output Creating temp dir: work_dirtry:
69dir_path = f"/tmp/{name}"70try:71 yield dir_path/tmp/work_dir72finally:with temp_directory("work_dir") as path:
70 try:71 yield dir_path/tmp/work_dir72 finally:73 print(f" Removing temp dir: {name}")7475print("\nTemporary directory:")7677with temp_directory("work_dir") as path:78 print(f" Using directory: {path/tmp/work_dir}")79 # Do work in temp directoryoutput Using directory: /tmp/work_dirfinally:
70try:71 yield dir_path72finally:73 print(f" Removing temp dir: {namework_dir}")output Removing temp dir: work_dir Removing temp dir: work_dirprint(" Database transactions:")
92print("\nDatabase transactions:")output Database transactions:def transaction(db_name):
pass 1 of 282@contextmanager83def transaction(db_nameusers_db):84 print(f" BEGIN TRANSACTION on {db_nameusers_db}")85 try:output BEGIN TRANSACTION on users_dbwith transaction("users_db"):
85 try:86 yield87 print(f" COMMIT on {db_nameusers_db}")88 except Exception as e:89 print(f" ROLLBACK on {db_name} ({e})")90 raise9192print("\nDatabase transactions:")9394with transaction("users_db"):95 print(" INSERT INTO users...")96 print(" UPDATE users...")output INSERT INTO users... UPDATE users... COMMIT on users_dbprint()
98print()def transaction(db_name):
pass 2 of 282@contextmanager83def transaction(db_nameorders_db):84 print(f" BEGIN TRANSACTION on {db_nameorders_db}")85 try:output BEGIN TRANSACTION on orders_dbwith transaction("orders_db"):
100try:101 with transaction("orders_db"):102 print(" INSERT INTO orders...")103 raise RuntimeError("Constraint violation")104except RuntimeError:output INSERT INTO orders...except Exception as e:
87 print(f" COMMIT on {db_name}")88except Exception as e:89 print(f" ROLLBACK on {db_nameorders_db} ({eConstraint violation})")90 raiseoutput ROLLBACK on orders_db (Constraint violation)except RuntimeError:
103 raise RuntimeError("Constraint violation")104except RuntimeError:105 print(" Transaction rolled back")106107# Changing contextoutput Transaction rolled back Transaction rolled backprint(" Working directory:")
117print("\nWorking directory:")output Working directory:original ← /current/dir
108@contextmanager109def working_directory(path/new/path):110 original→ /current/dir = "/current/dir"111 print(f" Changing to: {path/new/path}")112 try:output Changing to: /new/pathtry:
111print(f" Changing to: {path}")112try:113 yield path/new/path114finally:with working_directory("/new/path") as path:
112 try:113 yield path/new/path114 finally:115 print(f" Restoring to: {original}")116117print("\nWorking directory:")118119with working_directory("/new/path") as path:120 print(f" Working in: {path/new/path}")output Working in: /new/pathfinally:
112try:113 yield path114finally:115 print(f" Restoring to: {original/current/dir}")output Restoring to: /current/dir Restoring to: /current/dirmain()
122if __name__ == "__main__":123 main()
yield separates setup from cleanup. Code before yield runs on enter.
Multiple contexts
Manage several resources at once.
# Multiple context managers
from contextlib import contextmanager
def main():
# Nested with statements (old style)
@contextmanager
def resource(name):
print(f" Open: {name}")
yield name
print(f" Close: {name}")
print("Nested contexts (old style):\n")
with resource("Resource1"):
with resource("Resource2"):
with resource("Resource3"):
print(" Using all resources")
# Multiple contexts in one with (Python 3.1+)
print("\nMultiple in one with:")
with resource("R1"), resource("R2"), resource("R3"):
print(" Using all resources")
# File copy simulation
@contextmanager
def fake_file(name, mode):
print(f" Opening {name} ({mode})")
yield f"<{name} handle>"
print(f" Closing {name}")
print("\nFile copy:")
with fake_file("input.txt", "r") as src, fake_file("output.txt", "w") as dst:
print(f" Reading from {src}")
print(f" Writing to {dst}")
# Database connection and cursor
@contextmanager
def connection(db_name):
print(f" Connect to {db_name}")
yield f"<{db_name} connection>"
print(f" Disconnect from {db_name}")
@contextmanager
def cursor(conn):
print(f" Create cursor on {conn}")
yield f"<cursor>"
print(f" Close cursor")
print("\nDatabase operations:")
with connection("users_db") as conn:
with cursor(conn) as cur:
print(f" Execute query with {cur}")
# Cleanup order matters
class OrderedResource:
def __init__(self, name, order):
self.name = name
self.order = order
def __enter__(self):
print(f" [{self.order}] Enter: {self.name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f" [{self.order}] Exit: {self.name}")
return False
print("\nCleanup order:")
with OrderedResource("First", 1), \
OrderedResource("Second", 2), \
OrderedResource("Third", 3):
print(" [X] All acquired")
# Exception during acquisition
@contextmanager
def failing_resource(name, should_fail=False):
print(f" Acquiring {name}")
if should_fail:
raise RuntimeError(f"{name} failed to acquire")
yield name
print(f" Releasing {name}")
print("\nException during acquisition:")
try:
with failing_resource("R1"), \
failing_resource("R2", should_fail=True), \
failing_resource("R3"):
print(" Won't reach here")
except RuntimeError as e:
print(f" Error: {e}")
print(" Note: R1 was released, R3 never acquired")
# Nested transactions
@contextmanager
def savepoint(name):
print(f" SAVEPOINT {name}")
try:
yield
print(f" RELEASE SAVEPOINT {name}")
except Exception as e:
print(f" ROLLBACK TO SAVEPOINT {name}")
raise
print("\nNested transactions:")
try:
with savepoint("sp1"):
print(" Operation 1")
with savepoint("sp2"):
print(" Operation 2")
raise ValueError("Error in sp2")
except ValueError:
print(" Rolled back to sp1")
if __name__ == "__main__":
main()
def main(): # Nested with statements (old style)
5def main():6 # Nested with statements (old style)7 @contextmanager8 def resource(name):9 print(f" Open: {name}")10 yield name11 print(f" Close: {name}")12 13 print("Nested contexts (old style):\n")outputNested contexts (old style):def resource(name):
pass 1 of 67@contextmanager8def resource(nameResource1):9 print(f" Open: {nameResource1}")10 yield nameResource111 print(f" Close: {name}")output Open: Resource1All 6 passes — pass 1 is the card above pass name1 Resource1 2 Resource2 3 Resource1 4 R1 5 R2 6 R1 name ← Resource1
9 print(f" Open: {name}")10 yield name→ Resource111 print(f" Close: {nameResource1}")1213print("Nested contexts (old style):\n")1415with resource("Resource1"):16 with resource("Resource2"):17 with resource("Resource3"):18 print(" Using all resources")output Using all resources Close: Resource3 Close: Resource2 Close: Resource1print(" Multiple in one with:")
20# Multiple contexts in one with (Python 3.1+)21print("\nMultiple in one with:")output Multiple in one with:name ← R1
9 print(f" Open: {name}")10 yield name→ R111 print(f" Close: {nameR1}")1213print("Nested contexts (old style):\n")1415with resource("Resource1"):16 with resource("Resource2"):17 with resource("Resource3"):18 print(" Using all resources")1920# Multiple contexts in one with (Python 3.1+)21print("\nMultiple in one with:")2223with resource("R1"), resource("R2"), resource("R3"):24 print(" Using all resources")output Using all resources Close: R3 Close: R2 Close: R1print(" File copy:")
40print("\nFile copy:")output File copy:def fake_file(name, mode):
pass 1 of 234@contextmanager35def fake_file(nameinput.txt, moder):36 print(f" Opening {nameinput.txt} ({moder})")37 yield f"<{nameinput.txt} handle>"38 print(f" Closing {name}")output Opening input.txt (r)def fake_file(name, mode):
pass 2 of 234@contextmanager35def fake_file(nameoutput.txt, modew):36 print(f" Opening {nameoutput.txt} ({modew})")37 yield f"<{nameoutput.txt} handle>"38 print(f" Closing {name}")output Opening output.txt (w)name ← input.txt
36 print(f" Opening {name} ({mode})")37 yield f"<{name→ input.txt} handle>"38 print(f" Closing {nameinput.txt}")3940print("\nFile copy:")4142with fake_file("input.txt", "r") as src, fake_file("output.txt", "w") as dst:43 print(f" Reading from {src<input.txt handle>}")44 print(f" Writing to {dst<output.txt handle>}")output Reading from <input.txt handle> Writing to <output.txt handle> Closing output.txt Closing input.txtprint(" Database operations:")
59print("\nDatabase operations:")output Database operations:def connection(db_name):
47@contextmanager48def connection(db_nameusers_db):49 print(f" Connect to {db_nameusers_db}")50 yield f"<{db_nameusers_db} connection>"51 print(f" Disconnect from {db_name}")output Connect to users_dbdef cursor(conn):
53@contextmanager54def cursor(conn<users_db connection>):55 print(f" Create cursor on {conn<users_db connection>}")56 yield f"<cursor>"57 print(f" Close cursor")output Create cursor on <users_db connection>with cursor(conn) as cur:
49 print(f" Connect to {db_name}")50 yield f"<{db_nameusers_db} connection>"51 print(f" Disconnect from {db_nameusers_db}")5253@contextmanager54def cursor(conn):55 print(f" Create cursor on {conn}")56 yield f"<cursor>"57 print(f" Close cursor")5859print("\nDatabase operations:")6061with connection("users_db") as conn:62 with cursor(conn<users_db connection>) as cur:63 print(f" Execute query with {cur<cursor>}")output Execute query with <cursor> Close cursor Disconnect from users_db# Cleanup order matters
65# Cleanup order matters66class OrderedResource:67 def __init__(self, name, order):68 self.name = name69 self.order = order70 71 def __enter__(self):72 print(f" [{self.order}] Enter: {self.name}")73 return self74 75 def __exit__(self, exc_type, exc_val, exc_tb):76 print(f" [{self.order}] Exit: {self.name}")77 return False7879print("\nCleanup order:")output Cleanup order:self.name ← First, self.order ← 1
pass 1 of 366class OrderedResource:67 def __init__(self<__main__.main.<locals>.OrderedResource object at ⟨addr A⟩>, nameFirst, order1):68 self.name→ First = nameFirst69 self.order→ 1 = order1All 3 passes — pass 1 is the card above pass selfnameorderself.nameself.order1 <__main__.main.<locals>.OrderedResource object at ⟨addr A⟩> First 1 First 1 2 <__main__.main.<locals>.OrderedResource object at ⟨addr B⟩> Second 2 Second 2 3 <__main__.main.<locals>.OrderedResource object at ⟨addr C⟩> Third 3 Third 3 def __enter__(self):
pass 1 of 371def __enter__(self<__main__.main.<locals>.OrderedResource object at ⟨addr A⟩>):72 print(f" [{self.order1}] Enter: {self.nameFirst}")73 return selfoutput [1] Enter: FirstAll 3 passes — pass 1 is the card above pass selfself.orderself.name1 <__main__.main.<locals>.OrderedResource object at ⟨addr A⟩> 1 First 2 <__main__.main.<locals>.OrderedResource object at ⟨addr B⟩> 2 Second 3 <__main__.main.<locals>.OrderedResource object at ⟨addr C⟩> 3 Third with OrderedResource("First", 1), \ OrderedResource("Second",…
81with OrderedResource("First", 1), \82 OrderedResource("Second", 2), \83 OrderedResource("Third", 3):84 print(" [X] All acquired")output [X] All acquireddef __exit__(self, exc_type, exc_val, exc_tb):
pass 1 of 375def __exit__(self<__main__.main.<locals>.OrderedResource object at ⟨addr C⟩>, exc_typeNone, exc_valNone, exc_tbNone):76 print(f" [{self.order3}] Exit: {self.nameThird}")77 return Falseoutput [3] Exit: ThirdAll 3 passes — pass 1 is the card above pass selfself.orderself.name1 <__main__.main.<locals>.OrderedResource object at ⟨addr C⟩> 3 Third 2 <__main__.main.<locals>.OrderedResource object at ⟨addr B⟩> 2 Second 3 <__main__.main.<locals>.OrderedResource object at ⟨addr A⟩> 1 First print(" Exception during acquisition:")
95print("\nException during acquisition:")output Exception during acquisition:def failing_resource(name, should_fail=False):
pass 1 of 287@contextmanager88def failing_resource(nameR1, should_failFalse=FalseFalse):89 print(f" Acquiring {nameR1}")90 if should_fail:91 raise RuntimeError(f"{name} failed to acquire")92 yield nameR193 print(f" Releasing {name}")output Acquiring R1def failing_resource(name, should_fail=False):
pass 2 of 287@contextmanager88def failing_resource(nameR2, should_failTrue=FalseFalse):89 print(f" Acquiring {nameR2}")90 if should_fail:output Acquiring R2if should_fail:
89print(f" Acquiring {name}")90if should_failTrue:91 raise RuntimeError(f"{nameR2} failed to acquire")92yield nameexcept RuntimeError as e:
101 print(" Won't reach here")102except RuntimeError as e:103 print(f" Error: {eR2 failed to acquire}")104 print(" Note: R1 was released, R3 never acquired")105106# Nested transactionsoutput Error: R2 failed to acquire Note: R1 was released, R3 never acquired Note: R1 was released, R3 never acquiredprint(" Nested transactions:")
117print("\nNested transactions:")output Nested transactions:def savepoint(name):
pass 1 of 2107@contextmanager108def savepoint(namesp1):109 print(f" SAVEPOINT {namesp1}")110 try:output SAVEPOINT sp1with savepoint("sp1"):
119try:120 with savepoint("sp1"):121 print(" Operation 1")122 with savepoint("sp2"):output Operation 1def savepoint(name):
pass 2 of 2107@contextmanager108def savepoint(namesp2):109 print(f" SAVEPOINT {namesp2}")110 try:output SAVEPOINT sp2with savepoint("sp2"):
121 print(" Operation 1")122 with savepoint("sp2"):123 print(" Operation 2")124 raise ValueError("Error in sp2")125except ValueError:output Operation 2except Exception as e:
pass 1 of 2112 print(f" RELEASE SAVEPOINT {name}")113except Exception as e:114 print(f" ROLLBACK TO SAVEPOINT {namesp2}")115 raiseoutput ROLLBACK TO SAVEPOINT sp2except Exception as e:
pass 2 of 2112 print(f" RELEASE SAVEPOINT {name}")113except Exception as e:114 print(f" ROLLBACK TO SAVEPOINT {namesp1}")115 raiseoutput ROLLBACK TO SAVEPOINT sp1except ValueError:
124 raise ValueError("Error in sp2")125except ValueError:126 print(" Rolled back to sp1")output Rolled back to sp1 Rolled back to sp1main()
128if __name__ == "__main__":129 main()
with open(a) as f1, open(b) as f2: - all cleaned up properly.
contextlib utilities
Helpful functions for common patterns.
# contextlib utilities
from contextlib import contextmanager, suppress, redirect_stdout, closing
import io
def main():
# suppress - ignore specific exceptions
print("Using suppress:\n")
from contextlib import suppress
# Without suppress
try:
int("not a number")
except ValueError:
pass # Silently ignore
print(" Attempt 1: Failed silently")
# With suppress
with suppress(ValueError):
int("not a number")
print(" Attempt 2: Failed silently")
# Suppress multiple exception types
with suppress(ValueError, TypeError, KeyError):
d = {}
value = d["missing_key"]
print(" Attempt 3: Suppressed KeyError")
# redirect_stdout - capture print output
print("\nRedirecting stdout:")
output = io.StringIO()
with redirect_stdout(output):
print("This goes to StringIO")
print("Not to console")
captured = output.getvalue()
print(f" Captured: {repr(captured)}")
# closing - ensure object is closed
class Resource:
def __init__(self, name):
self.name = name
self.closed = False
def close(self):
print(f" Closing {self.name}")
self.closed = True
print("\nUsing closing:")
with closing(Resource("res1")) as res:
print(f" Using {res.name}")
print(f" Closed: {res.closed}")
# nullcontext - conditional context manager
from contextlib import nullcontext
print("\nConditional context:")
use_lock = False
@contextmanager
def lock():
print(" Acquiring lock")
yield
print(" Releasing lock")
context = lock() if use_lock else nullcontext()
with context:
print(" Critical section (no lock)")
use_lock = True
context = lock() if use_lock else nullcontext()
with context:
print(" Critical section (with lock)")
# ExitStack - dynamic context managers
from contextlib import ExitStack
print("\nExitStack:")
@contextmanager
def file_context(name):
print(f" Open {name}")
yield name
print(f" Close {name}")
files = ["file1.txt", "file2.txt", "file3.txt"]
with ExitStack() as stack:
handles = [stack.enter_context(file_context(f)) for f in files]
print(f" All files opened: {handles}")
print(" All files closed")
# ExitStack with callbacks
print("\nExitStack with callbacks:")
with ExitStack() as stack:
stack.callback(lambda: print(" Callback 1"))
stack.callback(lambda: print(" Callback 2"))
print(" In context")
# Practical: resource pool
print("\nResource pool:")
def acquire_resources(count):
stack = ExitStack()
resources = []
try:
for i in range(count):
res = Resource(f"res{i}")
stack.enter_context(closing(res))
resources.append(res)
return stack.pop_all(), resources
except:
stack.close()
raise
with ExitStack() as stack:
mgr, resources = acquire_resources(3)
stack.enter_context(mgr)
print(f" Using {len(resources)} resources")
if __name__ == "__main__":
main()
def main(): # suppress - ignore specific exceptions
6def main():7 # suppress - ignore specific exceptions8 print("Using suppress:\n")outputUsing suppress:print(" Attempt 1: Failed silently")
16 pass # Silently ignore17print(" Attempt 1: Failed silently")output Attempt 1: Failed silentlywith suppress(ValueError):
19# With suppress20with suppress(ValueError<class 'ValueError'>):21 int("not a number")22print(" Attempt 2: Failed silently")print(" Attempt 2: Failed silently")
21 int("not a number")22print(" Attempt 2: Failed silently")output Attempt 2: Failed silentlyd ← {}
24# Suppress multiple exception types25with suppress(ValueError<class 'ValueError'>, TypeError<class 'TypeError'>, KeyError<class 'KeyError'>):26 d→ {} = {}27 value = d["missing_key"](empty)28print(" Attempt 3: Suppressed KeyError")output ← ⟨StringIO A⟩, captured ← @@TRACE|contextlib_utilities.py|138704458339136|ENTER|41.5-42.1-43.32:with redirect_stdout(output):\n|41.26-41.32:output:39:⟨StringIO A⟩@@END @@TRACE|contextlib_utilities.py|138704458339136|BEFORE|42.9-42.39:print("This goes to StringIO")|@@END @@TRACE|contextlib_utilities.py|138704458339136|STDOUT|42.9-42.39:print("This goes to StringIO")|21:This goes to StringIO@@END @@TRACE|contextlib_utilities.py|138704458339136|AFTER|42.9-42.39:print("This goes to StringIO")|@@END @@TRACE|contextlib_utilities.py|138704458339136|BEFORE|43.9-43.32:print("Not to console")|@@END @@TRACE|contextlib_utilities.py|138704458339136|STDOUT|43.9-43.32:print("Not to console")|14:Not to console@@END @@TRACE|contextlib_utilities.py|138704458339136|AFTER|43.9-43.32:print("Not to console")|@@END
27 value = d["missing_key"]28print(" Attempt 3: Suppressed KeyError")2930#@help h131# suppress(*exceptions) catches and ignores specific exceptions32# cleaner than try-except-pass33# Use sparingly - explicit error handling is usually better34#@end3536# redirect_stdout - capture print output37print("\nRedirecting stdout:")3839output→ ⟨StringIO A⟩ = io<module 'io' (frozen)>.StringIO()4041with redirect_stdout(output):42 print("This goes to StringIO")43 print("Not to console")4445captured→ @@TRACE|contextlib_utilities.py|138704458339136|ENTER|41.5-42.1-43.32:with redirect_stdout(output):\n|41.26-41.32:output:39:⟨StringIO A⟩@@END @@TRACE|contextlib_utilities.py|138704458339136|BEFORE|42.9-42.39:print("This goes to StringIO")|@@END @@TRACE|contextlib_utilities.py|138704458339136|STDOUT|42.9-42.39:print("This goes to StringIO")|21:This goes to StringIO@@END @@TRACE|contextlib_utilities.py|138704458339136|AFTER|42.9-42.39:print("This goes to StringIO")|@@END @@TRACE|contextlib_utilities.py|138704458339136|BEFORE|43.9-43.32:print("Not to console")|@@END @@TRACE|contextlib_utilities.py|138704458339136|STDOUT|43.9-43.32:print("Not to console")|14:Not to console@@END @@TRACE|contextlib_utilities.py|138704458339136|AFTER|43.9-43.32:print("Not to console")|@@END = output⟨StringIO A⟩.getvalue()46print(f" Captured: {repr(captured@@TRACE|contextlib_utilities.py|138704458339136|ENTER|41.5-42.1-43.32:with redirect_stdout(output):\n|41.26-41.32:output:39:⟨StringIO A⟩@@END @@TRACE|contextlib_utilities.py|138704458339136|BEFORE|42.9-42.39:print("This goes to StringIO")|@@END @@TRACE|contextlib_utilities.py|138704458339136|STDOUT|42.9-42.39:print("This goes to StringIO")|21:This goes to StringIO@@END @@TRACE|contextlib_utilities.py|138704458339136|AFTER|42.9-42.39:print("This goes to StringIO")|@@END @@TRACE|contextlib_utilities.py|138704458339136|BEFORE|43.9-43.32:print("Not to console")|@@END @@TRACE|contextlib_utilities.py|138704458339136|STDOUT|43.9-43.32:print("Not to console")|14:Not to console@@END @@TRACE|contextlib_utilities.py|138704458339136|AFTER|43.9-43.32:print("Not to console")|@@END )}")4748# closing - ensure object is closed49class Resource:50 def __init__(self, name):51 self.name = name52 self.closed = False53 54 def close(self):55 print(f" Closing {self.name}")56 self.closed = True5758print("\nUsing closing:")output Attempt 3: Suppressed KeyError Redirecting stdout: Captured: '@@TRACE|contextlib_utilities.py|138704458339136|ENTER|41.5-42.1-43.32:with redirect_stdout(output):\\n|41.26-41.32:output:39:⟨StringIO A⟩@@END\n@@TRACE|contextlib_utilities.py|138704458339136|BEFORE|42.9-42.39:print("This goes to StringIO")|@@END\n@@TRACE|contextlib_utilities.py|138704458339136|STDOUT|42.9-42.39:print("This goes to StringIO")|21:This goes to StringIO@@END\n@@TRACE|contextlib_utilities.py|138704458339136|AFTER|42.9-42.39:print("This goes to StringIO")|@@END\n@@TRACE|contextlib_utilities.py|138704458339136|BEFORE|43.9-43.32:print("Not to console")|@@END\n@@TRACE|contextlib_utilities.py|138704458339136|STDOUT|43.9-43.32:print("Not to console")|14:Not to console@@END\n@@TRACE|contextlib_utilities.py|138704458339136|AFTER|43.9-43.32:print("Not to console")|@@END\n' Using closing:self.name ← res1, self.closed ← False
pass 1 of 449class Resource:50 def __init__(self<__main__.main.<locals>.Resource object at ⟨addr B⟩>, nameres1):51 self.name→ res1 = nameres152 self.closed→ False = FalseAll 4 passes — pass 1 is the card above pass selfnameres.nameself.nameself.closed1 <__main__.main.<locals>.Resource object at ⟨addr B⟩> res1 res1 res1 False 2 <__main__.main.<locals>.Resource object at ⟨addr C⟩> res0 — res0 False 3 <__main__.main.<locals>.Resource object at ⟨addr D⟩> res1 — res1 False 4 <__main__.main.<locals>.Resource object at ⟨addr E⟩> res2 — res2 False with closing(Resource("res1")) as res:
60with closing(Resource("res1")) as res:61 print(f" Using {res.nameres1}")62print(f" Closed: {res.closed}")output Using res1self.closed ← True
pass 1 of 454def close(self<__main__.main.<locals>.Resource object at ⟨addr B⟩>):55 print(f" Closing {self.nameres1}")56 self.closed→ True = Trueoutput Closing res1All 4 passes — pass 1 is the card above pass selfself.nameself.closed1 <__main__.main.<locals>.Resource object at ⟨addr B⟩> res1 True 2 <__main__.main.<locals>.Resource object at ⟨addr E⟩> res2 True 3 <__main__.main.<locals>.Resource object at ⟨addr D⟩> res1 True 4 <__main__.main.<locals>.Resource object at ⟨addr C⟩> res0 True use_lock ← False, context ← ⟨nullcontext F⟩
61 print(f" Using {res.name}")62print(f" Closed: {res.closedTrue}")6364# nullcontext - conditional context manager65from contextlib import nullcontext6667print("\nConditional context:")6869use_lock→ False = False7071@contextmanager72def lock():73 print(" Acquiring lock")74 yield75 print(" Releasing lock")7677context→ ⟨nullcontext F⟩ = lock() if use_lockFalse else nullcontext()output Closed: True Conditional context:with context:
79with context⟨nullcontext F⟩:80 print(" Critical section (no lock)")output Critical section (no lock)use_lock ← True, context ← ⟨_GeneratorContextManager G⟩
82use_lock→ True = True83context→ ⟨_GeneratorContextManager G⟩ = lock() if use_lockTrue else nullcontext()def lock():
71@contextmanager72def lock():73 print(" Acquiring lock")74 yield75 print(" Releasing lock")output Acquiring lockwith context:
73 print(" Acquiring lock")74 yield75 print(" Releasing lock")7677context = lock() if use_lock else nullcontext()7879with context:80 print(" Critical section (no lock)")8182use_lock = True83context = lock() if use_lock else nullcontext()8485with context⟨_GeneratorContextManager G⟩:86 print(" Critical section (with lock)")output Critical section (with lock) Releasing lockfiles ← ['file1.txt', 'file2.txt', 'file3.txt']
91print("\nExitStack:")9293@contextmanager94def file_context(name):95 print(f" Open {name}")96 yield name97 print(f" Close {name}")9899files→ ['file1.txt', 'file2.txt', 'file3.txt'] = ["file1.txt", "file2.txt", "file3.txt"]output ExitStack:with ExitStack() as stack:
101with ExitStack() as stack:102 handles = [stack⟨ExitStack F⟩.enter_context(file_context(f)) for f in files['file1.txt', 'file2.txt', 'file3.txt']]103 print(f" All files opened: {handles}")def file_context(name):
pass 1 of 393@contextmanager94def file_context(namefile1.txt):95 print(f" Open {namefile1.txt}")96 yield namefile1.txt97 print(f" Close {name}")output Open file1.txtAll 3 passes — pass 1 is the card above pass name1 file1.txt 2 file2.txt 3 file3.txt handles ← ['file1.txt', 'file2.txt', 'file3.txt'], name ← file1.txt
95 print(f" Open {name}")96 yield name→ file1.txt97 print(f" Close {namefile1.txt}")9899files = ["file1.txt", "file2.txt", "file3.txt"]100101with ExitStack() as stack:102 handles→ ['file1.txt', 'file2.txt', 'file3.txt'] = [stack⟨ExitStack F⟩.enter_context(file_context(f)) for f in files['file1.txt', 'file2.txt', 'file3.txt']]103 print(f" All files opened: {handles['file1.txt', 'file2.txt', 'file3.txt']}")output All files opened: ['file1.txt', 'file2.txt', 'file3.txt'] Close file3.txt Close file2.txt Close file1.txtprint(" All files closed")
105print(" All files closed")106107# ExitStack with callbacks108print("\nExitStack with callbacks:")output All files closed ExitStack with callbacks:with ExitStack() as stack:
110with ExitStack() as stack:111 stack⟨ExitStack H⟩.callback(lambda: print(" Callback 1"))112 stack⟨ExitStack H⟩.callback(lambda: print(" Callback 2"))113 print(" In context")output In contextprint(" Resource pool:")
115# Practical: resource pool116print("\nResource pool:")output Resource pool:stack ← ⟨ExitStack I⟩, resources ← []
118def acquire_resources(count3):119 stack→ ⟨ExitStack I⟩ = ExitStack()120 resources→ [] = []121 try:for i in range(count):
pass 1 of 3121try:122 for i0 in range(count3):123 res = Resource(f"res{i0}")124 stack.enter_context(closing(res))All 3 passes — pass 1 is the card above pass i1 0 2 1 3 2 res ← <__main__.main.<locals>.Resource object at ⟨addr C⟩>, resources ← [<__main__.main.<locals>.Resource object at ⟨addr C⟩>]
122for i in range(count):123 res→ <__main__.main.<locals>.Resource object at ⟨addr C⟩> = Resource(f"res{i0}")124 stack⟨ExitStack I⟩.enter_context(closing(res<__main__.main.<locals>.Resource object at ⟨addr C⟩>))125 resources→ [<__main__.main.<locals>.Resource object at ⟨addr C⟩>].append(res<__main__.main.<locals>.Resource object at ⟨addr C⟩>)126return stack.pop_all(), resourcesres ← <__main__.main.<locals>.Resource object at ⟨addr D⟩>, resources ← [<__main__.main.<locals>.Resource object at ⟨addr C⟩>, <__main__.main.<locals>.Resource object at ⟨addr D⟩>]
122for i in range(count):123 res→ <__main__.main.<locals>.Resource object at ⟨addr D⟩> = Resource(f"res{i1}")124 stack⟨ExitStack I⟩.enter_context(closing(res<__main__.main.<locals>.Resource object at ⟨addr D⟩>))125 resources→ [<__main__.main.<locals>.Resource object at ⟨addr C⟩>, <__main__.main.<locals>.Resource object at ⟨addr D⟩>].append(res<__main__.main.<locals>.Resource object at ⟨addr D⟩>)126return stack.pop_all(), resourcesres ← <__main__.main.<locals>.Resource object at ⟨addr E⟩>, resources ← [<__main__.main.<locals>.Resource object at ⟨addr C⟩>, <__main__.main.<locals>.Resource object at ⟨addr D⟩>, <__main__.main.<locals>.Resource object at ⟨addr E⟩>]
122for i in range(count):123 res→ <__main__.main.<locals>.Resource object at ⟨addr E⟩> = Resource(f"res{i2}")124 stack⟨ExitStack I⟩.enter_context(closing(res<__main__.main.<locals>.Resource object at ⟨addr E⟩>))125 resources→ [<__main__.main.<locals>.Resource object at ⟨addr C⟩>, <__main__.main.<locals>.Resource object at ⟨addr D⟩>, <__main__.main.<locals>.Resource object at ⟨addr E⟩>].append(res<__main__.main.<locals>.Resource object at ⟨addr E⟩>)126return stack.pop_all(), resourcesreturn stack.pop_all(), resources
125 resources.append(res)126 return stack⟨ExitStack I⟩.pop_all(), resources[<__main__.main.<locals>.Resource object at ⟨addr C⟩>, <__main__.main.<locals>.Resource object at ⟨addr D⟩>, <__main__.main.<locals>.Resource object at ⟨addr E⟩>]127except:mgr ← ⟨ExitStack J⟩, resources ← [<__main__.main.<locals>.Resource object at ⟨addr C⟩>, <__main__.main.<locals>.Resource object at ⟨addr D⟩>, <__main__.main.<locals>.Resource object at ⟨addr E⟩>]
131with ExitStack() as stack:132 mgr→ ⟨ExitStack J⟩, resources→ [<__main__.main.<locals>.Resource object at ⟨addr C⟩>, <__main__.main.<locals>.Resource object at ⟨addr D⟩>, <__main__.main.<locals>.Resource object at ⟨addr E⟩>] = acquire_resources(3)133 stack⟨ExitStack K⟩.enter_context(mgr⟨ExitStack J⟩)134 print(f" Using {len(resources[<__main__.main.<locals>.Resource object at ⟨addr C⟩>, <__main__.main.<locals>.Resource object at ⟨addr D⟩>, <__main__.main.<locals>.Resource object at ⟨addr E⟩>])} resources")output Using 3 resourcesmain()
136if __name__ == "__main__":137 main()
suppress(), redirect_stdout(), closing() - ready-made context managers.
Exercise: practical.py
Build a database connection context manager