# What happens behind the scenes

with something as variable:

do_something()

# Is equivalent to:

variable = something.__enter__()

try:

do_something()

finally:

something.__exit__(exc_type, exc_val, exc_tb)

class ManagedFile:

def __init__(self, filename, mode):

self.filename = filename

self.mode = mode

self.file = None

def __enter__(self):

print(f”Opening {self.filename}”)

self.file = open(self.filename, self.mode)

return self.file

def __exit__(self, exc_type, exc_val, exc_tb):

print(f”Closing {self.filename}”)

if self.file:

self.file.close()

# Return False to propagate exceptions (default)

return False

# Usage

with ManagedFile(“test.txt”, “w”) as f:

f.write(“Hello, World!”)

print(“Inside with block”)

# Output:

# Opening test.txt

# Inside with block

# Closing test.txt

class SuppressError:

def __enter__(self):

print(“Entering block”)

return self

def __exit__(self, exc_type, exc_val, exc_tb):

if exc_type is not None:

print(f”Suppressed error: {exc_val}”)

return True # Exception suppressed

print(“No error occurred”)

return False

with SuppressError():

print(“Doing something”)

raise ValueError(“Something went wrong”)

print(“This never runs”)

# Output:

# Entering block

# Doing something

# Suppressed error: Something went wrong

# Program continues (no crash)

from contextlib import contextmanager

@contextmanager

def managed_file(filename, mode):

print(f”Opening {filename}”)

f = open(filename, mode)

try:

yield f # The value that “as variable” receives

finally:

print(f”Closing {filename}”)

f.close()

# Usage (same as class-based version)

with managed_file(“test.txt”, “w”) as f:

f.write(“Hello from contextlib!”)

import sqlite3

from contextlib import contextmanager

@contextmanager

def get_db_connection(db_name):

conn = sqlite3.connect(db_name)

print(“Database connected”)

try:

yield conn

finally:

conn.close()

print(“Database disconnected”)

# Usage

with get_db_connection(“example.db”) as conn:

cursor = conn.cursor()

cursor.execute(“CREATE TABLE IF NOT EXISTS users (id INTEGER, name TEXT)”)

cursor.execute(“INSERT INTO users VALUES (1, ‘Ali’)”)

conn.commit()

print(“Database operations completed”)

# Connection is automatically closed here

import time

from contextlib import contextmanager

@contextmanager

def timer(name=”Operation”):

start = time.perf_counter()

print(f”{name} started…”)

try:

yield

finally:

end = time.perf_counter()

duration = end – start

print(f”{name} took {duration:.4f} seconds”)

# Usage

with timer(“Data processing”):

# Simulate some work

time.sleep(1)

total = sum(range(1000000))

print(f”Sum calculated: {total}”)

# Output:

# Data processing started…

# Sum calculated: 499999500000

# Data processing took 1.0234 seconds

import tempfile

import shutil

from pathlib import Path

from contextlib import contextmanager

@contextmanager

def temporary_directory():

temp_dir = tempfile.mkdtemp()

print(f”Created temporary directory: {temp_dir}”)

try:

yield Path(temp_dir)

finally:

shutil.rmtree(temp_dir)

print(“Temporary directory cleaned up”)

# Usage

with temporary_directory() as temp_dir:

# Create a file in the temp directory

file_path = temp_dir / “test.txt”

file_path.write_text(“Temporary content”)

print(f”File created at {file_path}”)

# Temp directory is deleted automatically

import sys

from io import StringIO

from contextlib import contextmanager

@contextmanager

def capture_output():

old_stdout = sys.stdout

captured = StringIO()

sys.stdout = captured

try:

yield captured

finally:

sys.stdout = old_stdout

# Usage

def noisy_function():

print(“This is printed”)

print(“This is also printed”)

return 42

with capture_output() as output:

result = noisy_function()

captured_text = output.getvalue()

print(f”Function returned: {result}”)

print(f”Captured output: {captured_text!r}”)

# 1. File handling (most common)

with open(“file.txt”, “r”) as f:

content = f.read()

# 2. Lock management (threading)

from threading import Lock

lock = Lock()

with lock:

# Critical section – lock is automatically released

pass

# 3. Temporary directories (Python 3.2+)

import tempfile

with tempfile.TemporaryDirectory() as tmpdir:

print(f”Working in {tmpdir}”)

# Directory is automatically deleted

# 4. Redirecting warnings

import warnings

with warnings.catch_warnings():

warnings.simplefilter(“ignore”)

# Warnings are suppressed here

pass

# 5. Changing directory temporarily (Python 3.11+)

import os

with os.chdir(“/tmp”):

print(f”Now in: {os.getcwd()}”)

print(f”Back to: {os.getcwd()}”)

# Nested explicitly

with open(“input.txt”, “r”) as infile:

with open(“output.txt”, “w”) as outfile:

outfile.write(infile.read())

# Multiple context managers in one line (Python 3.1+)

with open(“input.txt”, “r”) as infile, open(“output.txt”, “w”) as outfile:

outfile.write(infile.read())

# Custom context managers can also be nested

from contextlib import ExitStack

# Dynamic number of context managers

with ExitStack() as stack:

files = [stack.enter_context(open(f”file_{i}.txt”, “w”)) for i in range(5)]

for i, f in enumerate(files):

f.write(f”Content for file {i}”)

# All files are automatically closed

import time

from contextlib import contextmanager

from collections import deque

class ConnectionPool:

def __init__(self, size, create_connection_func):

self._pool = deque()

self._create_conn = create_connection_func

for _ in range(size):

self._pool.append(self._create_conn())

@contextmanager

def get_connection(self):

conn = self._pool.popleft()

print(f”Connection acquired ({len(self._pool)} left)”)

try:

yield conn

finally:

self._pool.append(conn)

print(f”Connection returned ({len(self._pool)} available)”)

# Simulated connection and usage

def create_mock_connection():

return {“id”: int(time.time() * 1000) % 10000}

pool = ConnectionPool(3, create_mock_connection)

with pool.get_connection() as conn1:

with pool.get_connection() as conn2:

with pool.get_connection() as conn3:

print(f”Using connections: {conn1[‘id’]}, {conn2[‘id’]}, {conn3[‘id’]}”)

# All connections are back in the pool