Bài 4: Context Managers

Mục Tiêu Bài Học

Sau khi hoàn thành bài này, bạn sẽ:

✅ Hiểu context managers và with statement
✅ Tạo context managers với __enter__ và __exit__
✅ Sử dụng contextlib module
✅ Áp dụng context managers cho resource management
✅ Tạo reusable context managers
✅ Xử lý exceptions trong context managers

Context Managers Là Gì?

Context Manager là object quản lý resources và đảm bảo cleanup code luôn được thực thi.

The Problem

# Without context manager - risky!file = open('data.txt', 'r')content = file.read()# What if exception happens here?file.close()  # Might never execute! # Better - but verbosefile = open('data.txt', 'r')try:    content = file.read()finally:    file.close()  # Always executes

The Solution - with Statement

# With context manager - clean and safe!with open('data.txt', 'r') as file:    content = file.read()# file.close() automatically called! # Multiple context managerswith open('input.txt', 'r') as infile, open('output.txt', 'w') as outfile:    content = infile.read()    outfile.write(content.upper())# Both files closed automatically

Context Manager Protocol

Context manager implement 2 methods: __enter__ và __exit__.

Basic Structure

class MyContextManager:    """Basic context manager."""        def __enter__(self):        """        Called when entering 'with' block.        Return value is assigned to 'as' variable.        """        print("Entering context")        return self  # Or any object        def __exit__(self, exc_type, exc_value, traceback):        """        Called when exiting 'with' block.                Args:            exc_type: Exception class (or None)            exc_value: Exception instance (or None)            traceback: Traceback object (or None)                Returns:            True: Suppress exception            False/None: Propagate exception        """        print("Exiting context")        return False  # Don't suppress exceptions # Usagewith MyContextManager() as cm:    print("Inside context")# Output:# Entering context# Inside context# Exiting context

How it Works

# This code:with context_manager as var:    # code block    pass # Is equivalent to:var = context_manager.__enter__()try:    # code block    passfinally:    context_manager.__exit__(None, None, None)

Custom Context Managers

1. File Handler

class FileHandler:    """Custom file context manager."""        def __init__(self, filename, mode='r'):        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_value, traceback):        print(f"Closing {self.filename}")        if self.file:            self.file.close()        return False # Usagewith FileHandler('test.txt', 'w') as f:    f.write("Hello, World!")# Opening test.txt# Closing test.txt # Verify file was writtenwith open('test.txt', 'r') as f:    print(f.read())  # Hello, World!

2. Timer Context Manager

import time class Timer:    """Measure execution time."""        def __init__(self, name="Operation"):        self.name = name        self.start_time = None        self.elapsed = None        def __enter__(self):        print(f"Starting {self.name}...")        self.start_time = time.time()        return self        def __exit__(self, exc_type, exc_value, traceback):        self.elapsed = time.time() - self.start_time        print(f"{self.name} took {self.elapsed:.4f} seconds")        return False # Usagewith Timer("Data processing"):    # Simulate work    time.sleep(1)    total = sum(range(1000000))# Starting Data processing...# Data processing took 1.0234 seconds # Access elapsed timewith Timer("Calculation") as timer:    result = sum(range(1000000)) print(f"Elapsed: {timer.elapsed:.4f}s")

3. Database Connection

class DatabaseConnection:    """Database connection context manager."""        def __init__(self, host, database):        self.host = host        self.database = database        self.connection = None        def __enter__(self):        print(f"Connecting to {self.database} on {self.host}")        # Simulate connection        self.connection = {            'host': self.host,            'database': self.database,            'connected': True        }        return self.connection        def __exit__(self, exc_type, exc_value, traceback):        print(f"Closing connection to {self.database}")        if self.connection:            self.connection['connected'] = False        return False # Usagewith DatabaseConnection('localhost', 'mydb') as conn:    print(f"Connected: {conn['connected']}")    # Execute queries...# Connecting to mydb on localhost# Connected: True# Closing connection to mydb

4. Directory Changer

import os class ChangeDirectory:    """Temporarily change working directory."""        def __init__(self, path):        self.path = path        self.old_path = None        def __enter__(self):        self.old_path = os.getcwd()        print(f"Changing directory: {self.old_path} -> {self.path}")        os.chdir(self.path)        return self.path        def __exit__(self, exc_type, exc_value, traceback):        print(f"Restoring directory: {self.path} -> {self.old_path}")        os.chdir(self.old_path)        return False # Usageprint(f"Current: {os.getcwd()}") with ChangeDirectory('/tmp'):    print(f"Inside with: {os.getcwd()}")    # Do work in /tmp print(f"After with: {os.getcwd()}")

5. Suppressing Exceptions

class SuppressException:    """Suppress specific exceptions."""        def __init__(self, *exceptions):        self.exceptions = exceptions        def __enter__(self):        return self        def __exit__(self, exc_type, exc_value, traceback):        # Return True to suppress exception        if exc_type is not None:            if issubclass(exc_type, self.exceptions):                print(f"Suppressed {exc_type.__name__}: {exc_value}")                return True  # Suppress        return False  # Don't suppress # Usageprint("Before") with SuppressException(ValueError, TypeError):    print("Inside")    raise ValueError("This error is suppressed")    print("This won't execute") print("After - execution continues!")# Before# Inside# Suppressed ValueError: This error is suppressed# After - execution continues! # Unsuppressed exceptiontry:    with SuppressException(ValueError):        raise KeyError("This will propagate")except KeyError as e:    print(f"Caught: {e}")

contextlib Module

Module contextlib cung cấp utilities để tạo context managers dễ hơn.

@contextmanager Decorator

from contextlib import contextmanager @contextmanagerdef file_handler(filename, mode='r'):    """Simple file context manager."""    print(f"Opening {filename}")    file = open(filename, mode)        try:        yield file  # Value returned to 'as' variable    finally:        print(f"Closing {filename}")        file.close() # Usagewith file_handler('test.txt', 'w') as f:    f.write("Using contextmanager decorator!")# Opening test.txt# Closing test.txt

Timer với @contextmanager

import timefrom contextlib import contextmanager @contextmanagerdef timer(name="Operation"):    """Timer context manager."""    print(f"Starting {name}...")    start = time.time()        try:        yield    finally:        elapsed = time.time() - start        print(f"{name} took {elapsed:.4f} seconds") # Usagewith timer("Processing"):    time.sleep(0.5)    result = sum(range(1000000))# Starting Processing...# Processing took 0.5234 seconds

Temporary State Change

from contextlib import contextmanager class Config:    """Simple configuration."""    debug = False @contextmanagerdef debug_mode():    """Temporarily enable debug mode."""    old_value = Config.debug    Config.debug = True    print("Debug mode: ON")        try:        yield    finally:        Config.debug = old_value        print("Debug mode: OFF") # Usageprint(f"Debug: {Config.debug}")  # False with debug_mode():    print(f"Debug: {Config.debug}")  # True    # Debug operations... print(f"Debug: {Config.debug}")  # False (restored)

suppress()

from contextlib import suppress # Suppress specific exceptionswith suppress(FileNotFoundError):    with open('nonexistent.txt', 'r') as f:        content = f.read()# No exception raised! print("Execution continues") # Multiple exceptionswith suppress(ValueError, TypeError, KeyError):    # Code that might raise these exceptions    result = int("not a number")# Silently suppressed

redirect_stdout() và redirect_stderr()

from contextlib import redirect_stdout, redirect_stderrimport io # Redirect stdoutoutput = io.StringIO() with redirect_stdout(output):    print("This goes to StringIO")    print("Not to console!") print("Back to console")print(f"Captured: {output.getvalue()}")# Back to console# Captured: This goes to StringIO# Not to console! # Redirect to filewith open('output.txt', 'w') as f:    with redirect_stdout(f):        print("Written to file")        print("Not to console") with open('output.txt', 'r') as f:    print(f"File content: {f.read()}")

ExitStack

from contextlib import ExitStack def process_files(filenames):    """Process multiple files with ExitStack."""    with ExitStack() as stack:        # Open all files        files = [stack.enter_context(open(fn, 'r')) for fn in filenames]                # Process files        for i, file in enumerate(files):            print(f"File {i}: {file.readline().strip()}")                # All files closed automatically # Create test filesfor i in range(3):    with open(f'file{i}.txt', 'w') as f:        f.write(f"Content of file {i}\n") # Processprocess_files(['file0.txt', 'file1.txt', 'file2.txt'])# File 0: Content of file 0# File 1: Content of file 1# File 2: Content of file 2

Conditional Context Manager

from contextlib import contextmanager, nullcontext @contextmanagerdef optional_context(condition, value):    """Context manager that's conditional."""    if condition:        print(f"Using context with {value}")        yield value    else:        yield None # Usagefor use_context in [True, False]:    ctx = optional_context(use_context, "data") if use_context else nullcontext()        with ctx:        print("Inside context")    print()

Real-world Examples

1. Database Transaction

from contextlib import contextmanager class Database:    """Database with transactions."""        def __init__(self):        self.in_transaction = False        self.data = {}        def begin(self):        self.in_transaction = True        self.backup = self.data.copy()        print("Transaction started")        def commit(self):        self.in_transaction = False        self.backup = None        print("Transaction committed")        def rollback(self):        self.data = self.backup.copy()        self.in_transaction = False        self.backup = None        print("Transaction rolled back")        @contextmanager    def transaction(self):        """Transaction context manager."""        self.begin()        try:            yield self            self.commit()        except Exception as e:            self.rollback()            raise # Usagedb = Database() # Successful transactionwith db.transaction():    db.data['user1'] = 'Alice'    db.data['user2'] = 'Bob'# Transaction started# Transaction committed print(db.data)  # {'user1': 'Alice', 'user2': 'Bob'} # Failed transactiontry:    with db.transaction():        db.data['user3'] = 'Charlie'        raise ValueError("Oops!")except ValueError:    pass# Transaction started# Transaction rolled back print(db.data)  # {'user1': 'Alice', 'user2': 'Bob'} (unchanged)

2. Lock Manager

import threadingfrom contextlib import contextmanager class Resource:    """Shared resource with locking."""        def __init__(self):        self.lock = threading.Lock()        self.value = 0        @contextmanager    def locked(self):        """Lock context manager."""        print(f"Thread {threading.current_thread().name}: Acquiring lock")        self.lock.acquire()                try:            yield self        finally:            print(f"Thread {threading.current_thread().name}: Releasing lock")            self.lock.release() # Usageresource = Resource() def worker(resource, iterations):    """Worker thread."""    for _ in range(iterations):        with resource.locked():            resource.value += 1 # Single-threaded demowith resource.locked():    resource.value += 10    print(f"Value: {resource.value}")# Thread MainThread: Acquiring lock# Value: 10# Thread MainThread: Releasing lock

3. Environment Variables

import osfrom contextlib import contextmanager @contextmanagerdef env_variable(key, value):    """Temporarily set environment variable."""    old_value = os.environ.get(key)        if value is None:        os.environ.pop(key, None)    else:        os.environ[key] = value        print(f"Set {key}={value}")        try:        yield    finally:        if old_value is None:            os.environ.pop(key, None)        else:            os.environ[key] = old_value        print(f"Restored {key}={old_value}") # Usageprint(f"Before: {os.environ.get('TEST_VAR', 'Not set')}") with env_variable('TEST_VAR', 'temporary_value'):    print(f"Inside: {os.environ.get('TEST_VAR')}") print(f"After: {os.environ.get('TEST_VAR', 'Not set')}")# Before: Not set# Set TEST_VAR=temporary_value# Inside: temporary_value# Restored TEST_VAR=None# After: Not set

4. Mock/Patch Context

from contextlib import contextmanager @contextmanagerdef mock_function(module, func_name, mock_impl):    """Mock a function temporarily."""    original = getattr(module, func_name)    setattr(module, func_name, mock_impl)    print(f"Mocked {module.__name__}.{func_name}")        try:        yield    finally:        setattr(module, func_name, original)        print(f"Restored {module.__name__}.{func_name}") # Example moduleclass MyModule:    @staticmethod    def get_data():        return "Real data" # Usageprint(MyModule.get_data())  # Real data with mock_function(MyModule, 'get_data', lambda: "Mock data"):    print(MyModule.get_data())  # Mock data print(MyModule.get_data())  # Real data (restored)

5. Profiler Context

import timefrom contextlib import contextmanager @contextmanagerdef profiler(name="Code block"):    """Profile code execution."""    stats = {        'name': name,        'start_time': None,        'end_time': None,        'elapsed': None    }        print(f"Profiling: {name}")    stats['start_time'] = time.time()        try:        yield stats    finally:        stats['end_time'] = time.time()        stats['elapsed'] = stats['end_time'] - stats['start_time']                print(f"\nProfile Results for '{name}':")        print(f"  Start: {stats['start_time']:.2f}")        print(f"  End: {stats['end_time']:.2f}")        print(f"  Elapsed: {stats['elapsed']:.4f}s") # Usagewith profiler("Data processing") as prof:    time.sleep(0.5)    result = sum(range(1000000))    time.sleep(0.3)# Profiling: Data processing## Profile Results for 'Data processing':#   Start: 1698400000.12#   End: 1698400000.92#   Elapsed: 0.8001s

Nested Context Managers

from contextlib import contextmanager @contextmanagerdef operation(name):    """Operation context."""    print(f"Start {name}")    try:        yield    finally:        print(f"End {name}") # Nested manuallywith operation("Outer"):    with operation("Middle"):        with operation("Inner"):            print("Core operation")# Start Outer# Start Middle# Start Inner# Core operation# End Inner# End Middle# End Outer # Multiple on one linewith operation("A"), operation("B"), operation("C"):    print("Core operation")

Best Practices

# 1. Always clean up in __exit__ or finally@contextmanagerdef resource_manager():    resource = acquire_resource()    try:        yield resource    finally:        release_resource(resource)  # Always executes # 2. Return False from __exit__ unless suppressingdef __exit__(self, exc_type, exc_value, traceback):    self.cleanup()    return False  # Let exceptions propagate # 3. Use contextlib for simple casesfrom contextlib import contextmanager @contextmanagerdef simple_context():    # Setup    try:        yield    finally:        # Cleanup        pass # 4. Document exception handling@contextmanagerdef documented_context():    """    Context manager that handles resources.        Exceptions are propagated after cleanup.    """    pass # 5. Use ExitStack for multiple resourcesfrom contextlib import ExitStack with ExitStack() as stack:    files = [stack.enter_context(open(f)) for f in filenames]    # All closed automatically

Bài Tập Thực Hành

Bài 1: Logging Context

Tạo context manager log entry/exit với indentation.

Bài 2: Cache Manager

Tạo context manager temporarily modify cache settings.

Bài 3: Retry Context

Tạo context manager retry operations on failure.

Bài 4: Atomic File Write

Tạo context manager write to temp file, rename on success.

Bài 5: Performance Monitor

Tạo context manager track memory và CPU usage.

Tóm Tắt

✅ Context Manager: Manage resources với setup/cleanup
✅ Protocol: __enter__() và __exit__()
✅ with statement: Automatic resource management
✅ @contextmanager: Decorator tạo context managers
✅ contextlib: Utilities như suppress, redirect, ExitStack
✅ Exception handling: Return True to suppress
✅ Real-world: Transactions, locks, environment, mocking

Bài Tiếp Theo

Bài 5: Advanced Functions - *args, **kwargs, partial, closures, và function attributes! 🚀

Remember:

Always clean up resources
Use contextlib for simple cases
Return False to propagate exceptions
ExitStack for multiple resources
Context managers = clean code! 🎯

#resource-management

Context Managers