1"""
2Python Context Managers
3
4Demonstrates:
5- with statement and context protocol
6- __enter__ and __exit__ methods
7- contextlib.contextmanager decorator
8- contextlib.suppress
9- contextlib.ExitStack
10- Practical examples (timer, file transaction, resource management)
11"""
12
13import contextlib
14import time
15from typing import Any, Optional
16
17
18def section(title: str) -> None:
19 """Print a section header."""
20 print("\n" + "=" * 60)
21 print(f" {title}")
22 print("=" * 60)
23
24
25# =============================================================================
26# Basic Context Manager Protocol
27# =============================================================================
28
29section("Basic Context Manager Protocol")
30
31
32class Timer:
33 """Context manager for timing code blocks."""
34
35 def __init__(self, name: str = "block"):
36 self.name = name
37 self.start_time: Optional[float] = None
38 self.elapsed: Optional[float] = None
39
40 def __enter__(self):
41 """Called when entering 'with' block."""
42 print(f" [{self.name}] Starting timer")
43 self.start_time = time.perf_counter()
44 return self
45
46 def __exit__(self, exc_type, exc_val, exc_tb):
47 """Called when exiting 'with' block."""
48 end_time = time.perf_counter()
49 self.elapsed = end_time - self.start_time
50 print(f" [{self.name}] Elapsed: {self.elapsed * 1000:.4f} ms")
51
52 # Return False to propagate exceptions, True to suppress
53 return False
54
55
56with Timer("computation") as timer:
57 print(" Doing some work...")
58 time.sleep(0.1)
59 result = sum(range(100000))
60
61print(f"Timer object elapsed: {timer.elapsed * 1000:.4f} ms")
62
63
64# =============================================================================
65# Resource Management
66# =============================================================================
67
68section("Resource Management")
69
70
71class ManagedFile:
72 """Context manager for file-like resource."""
73
74 def __init__(self, filename: str):
75 self.filename = filename
76 self.file = None
77 print(f" Initializing ManagedFile({filename})")
78
79 def __enter__(self):
80 print(f" Opening {self.filename}")
81 self.file = open(self.filename, 'w')
82 return self.file
83
84 def __exit__(self, exc_type, exc_val, exc_tb):
85 if self.file:
86 print(f" Closing {self.filename}")
87 self.file.close()
88
89 if exc_type is not None:
90 print(f" Exception occurred: {exc_type.__name__}: {exc_val}")
91
92 return False # Don't suppress exceptions
93
94
95# Temporary file for demo
96import tempfile
97import os
98
99temp_path = os.path.join(tempfile.gettempdir(), "demo.txt")
100
101with ManagedFile(temp_path) as f:
102 f.write("Hello, context managers!\n")
103 print(f" Written to {temp_path}")
104
105print(f"File closed automatically")
106
107
108# =============================================================================
109# contextlib.contextmanager Decorator
110# =============================================================================
111
112section("contextlib.contextmanager Decorator")
113
114
115@contextlib.contextmanager
116def simple_timer(name: str):
117 """Simpler timer using contextmanager decorator."""
118 print(f" [{name}] Starting...")
119 start = time.perf_counter()
120
121 try:
122 yield # Control returns to with block here
123 finally:
124 # This runs when exiting with block
125 elapsed = time.perf_counter() - start
126 print(f" [{name}] Finished in {elapsed * 1000:.4f} ms")
127
128
129with simple_timer("generator-based"):
130 print(" Executing code block...")
131 time.sleep(0.05)
132
133
134@contextlib.contextmanager
135def atomic_write(filename: str):
136 """Atomic file write - write to temp, then rename."""
137 temp_file = filename + ".tmp"
138
139 print(f" Writing to temporary file: {temp_file}")
140 f = open(temp_file, 'w')
141
142 try:
143 yield f
144 except Exception:
145 print(f" Error occurred, removing temp file")
146 f.close()
147 os.remove(temp_file)
148 raise
149 else:
150 print(f" Success, committing to {filename}")
151 f.close()
152 os.rename(temp_file, filename)
153
154
155atomic_path = os.path.join(tempfile.gettempdir(), "atomic.txt")
156
157with atomic_write(atomic_path) as f:
158 f.write("Data written atomically\n")
159 print(" Writing data...")
160
161print(f"File committed successfully")
162
163
164# =============================================================================
165# contextlib.suppress
166# =============================================================================
167
168section("contextlib.suppress")
169
170print("Without suppress:")
171try:
172 with open("/nonexistent/file.txt") as f:
173 pass
174except FileNotFoundError as e:
175 print(f" Caught exception: {e.__class__.__name__}")
176
177print("\nWith suppress:")
178with contextlib.suppress(FileNotFoundError):
179 with open("/nonexistent/file.txt") as f:
180 pass
181 print(" This won't print")
182
183print(" No exception raised, execution continues")
184
185
186# =============================================================================
187# Nested Context Managers
188# =============================================================================
189
190section("Nested Context Managers")
191
192
193@contextlib.contextmanager
194def tag(name: str):
195 """HTML tag context manager."""
196 print(f"<{name}>", end="")
197 yield
198 print(f"</{name}>", end="")
199
200
201print("Manual nesting:")
202with tag("html"):
203 with tag("body"):
204 with tag("h1"):
205 print("Hello", end="")
206print()
207
208
209# =============================================================================
210# contextlib.ExitStack
211# =============================================================================
212
213section("contextlib.ExitStack")
214
215
216@contextlib.contextmanager
217def resource(name: str):
218 """Simulate acquiring a resource."""
219 print(f" Acquiring {name}")
220 yield name
221 print(f" Releasing {name}")
222
223
224# Dynamically manage multiple context managers
225with contextlib.ExitStack() as stack:
226 resources = []
227 for i in range(3):
228 r = stack.enter_context(resource(f"Resource-{i}"))
229 resources.append(r)
230
231 print(f" All resources acquired: {resources}")
232 print(f" Using resources...")
233
234print(" All resources released in reverse order")
235
236
237# =============================================================================
238# Transaction Pattern
239# =============================================================================
240
241section("Transaction Pattern")
242
243
244class Transaction:
245 """Database-like transaction context manager."""
246
247 def __init__(self, name: str):
248 self.name = name
249 self.committed = False
250
251 def __enter__(self):
252 print(f" BEGIN TRANSACTION '{self.name}'")
253 return self
254
255 def __exit__(self, exc_type, exc_val, exc_tb):
256 if exc_type is None and self.committed:
257 print(f" COMMIT '{self.name}'")
258 else:
259 print(f" ROLLBACK '{self.name}'")
260 if exc_type:
261 print(f" Reason: {exc_type.__name__}: {exc_val}")
262 return False
263
264 def commit(self):
265 """Mark transaction for commit."""
266 self.committed = True
267
268
269print("Successful transaction:")
270with Transaction("transfer-funds") as txn:
271 print(" Deducting from account A")
272 print(" Adding to account B")
273 txn.commit()
274
275print("\nFailed transaction:")
276try:
277 with Transaction("invalid-transfer") as txn:
278 print(" Deducting from account A")
279 raise ValueError("Insufficient funds")
280 txn.commit() # Never reached
281except ValueError:
282 pass
283
284
285# =============================================================================
286# Reentrant Context Manager
287# =============================================================================
288
289section("Reentrant Context Manager")
290
291
292class ReentrantResource:
293 """Context manager that can be entered multiple times."""
294
295 def __init__(self, name: str):
296 self.name = name
297 self.level = 0
298
299 def __enter__(self):
300 self.level += 1
301 print(f" [{self.name}] Enter level {self.level}")
302 return self
303
304 def __exit__(self, exc_type, exc_val, exc_tb):
305 print(f" [{self.name}] Exit level {self.level}")
306 self.level -= 1
307 return False
308
309
310resource = ReentrantResource("lock")
311
312with resource:
313 print(" Outer block")
314 with resource:
315 print(" Inner block")
316 with resource:
317 print(" Innermost block")
318
319
320# =============================================================================
321# Cleanup Actions
322# =============================================================================
323
324section("Cleanup Actions with ExitStack.callback")
325
326
327with contextlib.ExitStack() as stack:
328 # Register cleanup callbacks
329 stack.callback(lambda: print(" Cleanup 1"))
330 stack.callback(lambda: print(" Cleanup 2"))
331 stack.callback(lambda: print(" Cleanup 3"))
332
333 print(" Main code executing...")
334
335print(" Callbacks executed in LIFO order")
336
337
338# =============================================================================
339# Summary
340# =============================================================================
341
342section("Summary")
343
344print("""
345Context manager patterns covered:
3461. __enter__/__exit__ protocol - manual implementation
3472. contextlib.contextmanager - decorator-based approach
3483. contextlib.suppress - ignore specific exceptions
3494. contextlib.ExitStack - dynamic context manager composition
3505. Timer pattern - measure execution time
3516. Transaction pattern - commit/rollback semantics
3527. Resource management - automatic cleanup
3538. Atomic operations - all-or-nothing file writes
354
355Context managers ensure proper resource cleanup and
356provide clean syntax for setup/teardown operations.
357""")
358
359# Cleanup temp files
360try:
361 os.remove(temp_path)
362 os.remove(atomic_path)
363except:
364 pass