Decorators & Context Managers

 1def timing_decorator(func):
 2    """A decorator that measures function execution time."""
 3    import time
 4    
 5    def wrapper(*args, **kwargs):
 6        start_time = time.time()
 7        result = func(*args, **kwargs)
 8        end_time = time.time()
 9        print(f"{func.__name__} took {end_time - start_time:.4f} seconds")
10        return result
11    
12    return wrapper
13
14@timing_decorator
15def slow_function():
16    """A function that takes some time to execute."""
17    import time
18    time.sleep(1)
19    return "Task completed"
20
21# Usage
22result = slow_function()
23# Output: slow_function took 1.0041 seconds
24print(result)  # Output: Task completed

 1from functools import wraps
 2
 3def logging_decorator(func):
 4    """A decorator that logs function calls and preserves metadata."""
 5    
 6    @wraps(func)  # Preserves original function's metadata
 7    def wrapper(*args, **kwargs):
 8        print(f"Calling {func.__name__} with args={args}, kwargs={kwargs}")
 9        try:
10            result = func(*args, **kwargs)
11            print(f"{func.__name__} returned: {result}")
12            return result
13        except Exception as e:
14            print(f"{func.__name__} raised: {e}")
15            raise
16    
17    return wrapper
18
19@logging_decorator
20def divide(a, b):
21    """Divide two numbers."""
22    return a / b
23
24# Usage
25print(f"Function name: {divide.__name__}")  # Output: divide (not wrapper)
26print(f"Function doc: {divide.__doc__}")    # Output: Divide two numbers.
27
28result = divide(10, 2)
29# Output: Calling divide with args=(10, 2), kwargs={}
30# Output: divide returned: 5.0
31
32try:
33    divide(10, 0)
34except ZeroDivisionError:
35    print("Caught division by zero")
36# Output: Calling divide with args=(10, 0), kwargs={}
37# Output: divide raised: division by zero

 1def retry(max_attempts=3, delay=1):
 2    """Decorator that retries a function on failure."""
 3    def decorator(func):
 4        @wraps(func)
 5        def wrapper(*args, **kwargs):
 6            for attempt in range(max_attempts):
 7                try:
 8                    return func(*args, **kwargs)
 9                except Exception as e:
10                    if attempt == max_attempts - 1:
11                        print(f"Failed after {max_attempts} attempts")
12                        raise
13                    print(f"Attempt {attempt + 1} failed: {e}. Retrying...")
14                    time.sleep(delay)
15        return wrapper
16    return decorator
17
18@retry(max_attempts=3, delay=0.5)
19def unreliable_network_call():
20    """Simulates an unreliable network operation."""
21    import random
22    if random.random() < 0.7:  # 70% chance of failure
23        raise ConnectionError("Network timeout")
24    return "Success!"
25
26# Usage
27try:
28    result = unreliable_network_call()
29    print(result)
30except ConnectionError:
31    print("All attempts failed")

 1def memoize(maxsize=128):
 2    """A simple memoization decorator with size limit."""
 3    def decorator(func):
 4        cache = {}
 5        cache_order = []
 6        
 7        @wraps(func)
 8        def wrapper(*args, **kwargs):
 9            # Create a hashable key
10            key = str(args) + str(sorted(kwargs.items()))
11            
12            if key in cache:
13                print(f"Cache hit for {func.__name__}{args}")
14                return cache[key]
15            
16            # Compute result
17            result = func(*args, **kwargs)
18            print(f"Cache miss for {func.__name__}{args}, computing...")
19            
20            # Manage cache size
21            if len(cache) >= maxsize:
22                oldest_key = cache_order.pop(0)
23                del cache[oldest_key]
24            
25            cache[key] = result
26            cache_order.append(key)
27            return result
28        
29        # Add cache inspection methods
30        wrapper.cache_info = lambda: {
31            'size': len(cache),
32            'maxsize': maxsize,
33            'keys': list(cache.keys())
34        }
35        wrapper.cache_clear = lambda: cache.clear() or cache_order.clear()
36        
37        return wrapper
38    return decorator
39
40@memoize(maxsize=5)
41def fibonacci(n):
42    """Calculate Fibonacci number (inefficient recursive version)."""
43    if n < 2:
44        return n
45    return fibonacci(n - 1) + fibonacci(n - 2)
46
47# Usage
48print(fibonacci(10))  # Computes and caches intermediate results
49print(fibonacci(10))  # Cache hit!
50print(fibonacci.cache_info())  # Inspect cache state

 1import time
 2from collections import defaultdict
 3
 4class RateLimiter:
 5    """A class-based decorator for rate limiting function calls."""
 6    
 7    def __init__(self, max_calls=5, time_window=60):
 8        self.max_calls = max_calls
 9        self.time_window = time_window
10        self.calls = defaultdict(list)  # Track calls per function
11    
12    def __call__(self, func):
13        @wraps(func)
14        def wrapper(*args, **kwargs):
15            now = time.time()
16            func_calls = self.calls[func.__name__]
17            
18            # Remove old calls outside the time window
19            func_calls[:] = [call_time for call_time in func_calls 
20                           if now - call_time < self.time_window]
21            
22            # Check rate limit
23            if len(func_calls) >= self.max_calls:
24                raise Exception(f"Rate limit exceeded for {func.__name__}. "
25                              f"Max {self.max_calls} calls per {self.time_window}s")
26            
27            # Record this call and execute function
28            func_calls.append(now)
29            return func(*args, **kwargs)
30        
31        return wrapper
32
33@RateLimiter(max_calls=3, time_window=10)
34def api_call(endpoint):
35    """Simulate an API call."""
36    print(f"Making API call to {endpoint}")
37    return f"Response from {endpoint}"
38
39# Usage
40for i in range(5):
41    try:
42        result = api_call(f"/endpoint_{i}")
43        print(result)
44        time.sleep(1)
45    except Exception as e:
46        print(f"Error: {e}")
47        break

 1class ValidatedProperty:
 2    """A descriptor that validates property values."""
 3    
 4    def __init__(self, validator=None, doc=None):
 5        self.validator = validator
 6        self.doc = doc
 7        self.name = None  # Set by __set_name__
 8    
 9    def __set_name__(self, owner, name):
10        self.name = name
11        self.private_name = f'_{name}'
12    
13    def __get__(self, instance, owner):
14        if instance is None:
15            return self
16        return getattr(instance, self.private_name, None)
17    
18    def __set__(self, instance, value):
19        if self.validator:
20            value = self.validator(value)
21        setattr(instance, self.private_name, value)
22
23def positive_number(value):
24    """Validator for positive numbers."""
25    if not isinstance(value, (int, float)):
26        raise TypeError("Must be a number")
27    if value <= 0:
28        raise ValueError("Must be positive")
29    return value
30
31def non_empty_string(value):
32    """Validator for non-empty strings."""
33    if not isinstance(value, str):
34        raise TypeError("Must be a string")
35    if not value.strip():
36        raise ValueError("Must not be empty")
37    return value.strip()
38
39class Product:
40    """A product with validated properties."""
41    
42    price = ValidatedProperty(positive_number, "Product price")
43    name = ValidatedProperty(non_empty_string, "Product name")
44    
45    def __init__(self, name, price):
46        self.name = name
47        self.price = price
48    
49    def __repr__(self):
50        return f"Product(name='{self.name}', price={self.price})"
51
52# Usage
53try:
54    product = Product("Widget", 19.99)
55    print(product)  # Product(name='Widget', price=19.99)
56    
57    product.price = 29.99  # Valid update
58    print(f"Updated price: {product.price}")
59    
60    product.price = -5  # Raises ValueError
61except ValueError as e:
62    print(f"Validation error: {e}")

 1# File handling with automatic cleanup
 2def read_config_file(filename):
 3    """Read configuration from a file safely."""
 4    config = {}
 5    
 6    try:
 7        with open(filename, 'r') as file:
 8            for line_num, line in enumerate(file, 1):
 9                line = line.strip()
10                if line and not line.startswith('#'):
11                    try:
12                        key, value = line.split('=', 1)
13                        config[key.strip()] = value.strip()
14                    except ValueError:
15                        print(f"Invalid line {line_num}: {line}")
16    except FileNotFoundError:
17        print(f"Config file {filename} not found, using defaults")
18    except PermissionError:
19        print(f"Permission denied reading {filename}")
20    
21    return config
22
23# Usage
24config = read_config_file('app.conf')
25print(config)
26# File is automatically closed even if an exception occurs

 1def copy_file_with_backup(source, destination, backup_suffix='.bak'):
 2    """Copy a file while creating a backup of the destination."""
 3    import shutil
 4    import os
 5    
 6    backup_path = destination + backup_suffix
 7    
 8    try:
 9        # Handle multiple files with nested context managers
10        with open(source, 'rb') as src, \
11             open(destination, 'rb') as dst_read:
12            
13            # Create backup first
14            with open(backup_path, 'wb') as backup:
15                shutil.copyfileobj(dst_read, backup)
16            
17            print(f"Backup created: {backup_path}")
18        
19        # Now perform the actual copy
20        with open(source, 'rb') as src, \
21             open(destination, 'wb') as dst:
22            shutil.copyfileobj(src, dst)
23            
24        print(f"File copied: {source} -> {destination}")
25        
26    except FileNotFoundError as e:
27        print(f"File not found: {e}")
28    except PermissionError as e:
29        print(f"Permission error: {e}")
30    except Exception as e:
31        print(f"Unexpected error: {e}")
32        # Clean up backup if copy failed
33        if os.path.exists(backup_path):
34            os.remove(backup_path)
35            print("Backup removed due to copy failure")
36
37# Usage
38copy_file_with_backup('source.txt', 'destination.txt')

  1import time
  2import threading
  3
  4class TimingContext:
  5    """Context manager for measuring execution time."""
  6    
  7    def __init__(self, description="Operation", verbose=True):
  8        self.description = description
  9        self.verbose = verbose
 10        self.start_time = None
 11        self.end_time = None
 12    
 13    def __enter__(self):
 14        self.start_time = time.time()
 15        if self.verbose:
 16            print(f"Starting {self.description}...")
 17        return self
 18    
 19    def __exit__(self, exc_type, exc_val, exc_tb):
 20        self.end_time = time.time()
 21        duration = self.end_time - self.start_time
 22        
 23        if exc_type is not None:
 24            print(f"{self.description} failed after {duration:.4f} seconds")
 25            return False  # Don't suppress the exception
 26        
 27        if self.verbose:
 28            print(f"{self.description} completed in {duration:.4f} seconds")
 29        
 30        return False
 31    
 32    @property
 33    def duration(self):
 34        """Get the duration of the operation."""
 35        if self.start_time and self.end_time:
 36            return self.end_time - self.start_time
 37        return None
 38
 39class DatabaseConnection:
 40    """Mock database connection with transaction support."""
 41    
 42    def __init__(self, connection_string):
 43        self.connection_string = connection_string
 44        self.connected = False
 45        self.transaction_active = False
 46    
 47    def connect(self):
 48        print(f"Connecting to {self.connection_string}")
 49        time.sleep(0.1)  # Simulate connection time
 50        self.connected = True
 51    
 52    def disconnect(self):
 53        if self.connected:
 54            print("Disconnecting from database")
 55            self.connected = False
 56    
 57    def begin_transaction(self):
 58        if not self.connected:
 59            raise Exception("Not connected to database")
 60        print("Beginning transaction")
 61        self.transaction_active = True
 62    
 63    def commit(self):
 64        if self.transaction_active:
 65            print("Committing transaction")
 66            self.transaction_active = False
 67    
 68    def rollback(self):
 69        if self.transaction_active:
 70            print("Rolling back transaction")
 71            self.transaction_active = False
 72    
 73    def execute(self, query):
 74        if not self.connected:
 75            raise Exception("Not connected to database")
 76        print(f"Executing: {query}")
 77        if "ERROR" in query:
 78            raise Exception("SQL Error")
 79
 80class DatabaseTransaction:
 81    """Context manager for database transactions."""
 82    
 83    def __init__(self, connection):
 84        self.connection = connection
 85    
 86    def __enter__(self):
 87        self.connection.begin_transaction()
 88        return self.connection
 89    
 90    def __exit__(self, exc_type, exc_val, exc_tb):
 91        if exc_type is not None:
 92            self.connection.rollback()
 93            print(f"Transaction rolled back due to: {exc_val}")
 94            return False  # Don't suppress the exception
 95        else:
 96            self.connection.commit()
 97        return False
 98
 99# Usage
100db = DatabaseConnection("postgresql://localhost/mydb")
101db.connect()
102
103try:
104    with TimingContext("Database operations"):
105        with DatabaseTransaction(db) as conn:
106            conn.execute("INSERT INTO users (name) VALUES ('Alice')")
107            conn.execute("INSERT INTO users (name) VALUES ('Bob')")
108            # Transaction automatically commits
109            
110    print("All operations successful")
111    
112except Exception as e:
113    print(f"Operation failed: {e}")
114finally:
115    db.disconnect()

  1from contextlib import contextmanager
  2import threading
  3import os
  4import tempfile
  5
  6@contextmanager
  7def temporary_directory():
  8    """Create a temporary directory that's cleaned up automatically."""
  9    temp_dir = tempfile.mkdtemp()
 10    try:
 11        print(f"Created temporary directory: {temp_dir}")
 12        yield temp_dir
 13    finally:
 14        import shutil
 15        shutil.rmtree(temp_dir)
 16        print(f"Cleaned up temporary directory: {temp_dir}")
 17
 18@contextmanager
 19def thread_lock(lock, timeout=None):
 20    """Context manager for thread locks with timeout."""
 21    acquired = lock.acquire(timeout=timeout)
 22    if not acquired:
 23        raise TimeoutError(f"Could not acquire lock within {timeout} seconds")
 24    
 25    try:
 26        yield
 27    finally:
 28        lock.release()
 29
 30@contextmanager
 31def suppress_stdout():
 32    """Temporarily suppress stdout output."""
 33    import sys
 34    from io import StringIO
 35    
 36    old_stdout = sys.stdout
 37    sys.stdout = StringIO()
 38    try:
 39        yield
 40    finally:
 41        sys.stdout = old_stdout
 42
 43@contextmanager
 44def environment_variable(name, value):
 45    """Temporarily set an environment variable."""
 46    old_value = os.environ.get(name)
 47    os.environ[name] = value
 48    try:
 49        yield
 50    finally:
 51        if old_value is None:
 52            os.environ.pop(name, None)
 53        else:
 54            os.environ[name] = old_value
 55
 56# Usage examples
 57def demonstrate_context_managers():
 58    """Demonstrate various custom context managers."""
 59    
 60    # Temporary directory
 61    with temporary_directory() as temp_dir:
 62        test_file = os.path.join(temp_dir, "test.txt")
 63        with open(test_file, 'w') as f:
 64            f.write("This is a test file")
 65        print(f"Created file: {test_file}")
 66        print(f"File exists: {os.path.exists(test_file)}")
 67    # Directory and file are automatically cleaned up
 68    
 69    # Thread synchronization
 70    lock = threading.Lock()
 71    
 72    def worker(worker_id):
 73        try:
 74            with thread_lock(lock, timeout=1.0):
 75                print(f"Worker {worker_id} acquired lock")
 76                time.sleep(0.5)
 77                print(f"Worker {worker_id} releasing lock")
 78        except TimeoutError:
 79            print(f"Worker {worker_id} timed out waiting for lock")
 80    
 81    # Start multiple threads
 82    threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
 83    for t in threads:
 84        t.start()
 85    for t in threads:
 86        t.join()
 87    
 88    # Suppress output
 89    print("This will be printed")
 90    with suppress_stdout():
 91        print("This will not be printed")
 92    print("This will be printed again")
 93    
 94    # Environment variable
 95    print(f"HOME before: {os.environ.get('TEST_VAR', 'Not set')}")
 96    with environment_variable('TEST_VAR', 'temporary_value'):
 97        print(f"HOME during: {os.environ.get('TEST_VAR')}")
 98    print(f"HOME after: {os.environ.get('TEST_VAR', 'Not set')}")
 99
100demonstrate_context_managers()

 1from contextlib import contextmanager
 2import logging
 3import functools
 4
 5# Set up logging
 6logging.basicConfig(level=logging.INFO)
 7logger = logging.getLogger(__name__)
 8
 9@contextmanager
10def performance_monitor(operation_name):
11    """Context manager that monitors performance and logs results."""
12    import psutil
13    import time
14    
15    process = psutil.Process()
16    start_time = time.time()
17    start_memory = process.memory_info().rss / 1024 / 1024  # MB
18    start_cpu = process.cpu_percent()
19    
20    logger.info(f"Starting {operation_name}")
21    
22    try:
23        yield
24    finally:
25        end_time = time.time()
26        end_memory = process.memory_info().rss / 1024 / 1024  # MB
27        end_cpu = process.cpu_percent()
28        
29        duration = end_time - start_time
30        memory_diff = end_memory - start_memory
31        
32        logger.info(
33            f"Completed {operation_name}: "
34            f"Duration={duration:.2f}s, "
35            f"Memory={memory_diff:+.2f}MB, "
36            f"CPU={end_cpu:.1f}%"
37        )
38
39def monitor_performance(operation_name=None):
40    """Decorator that monitors function performance."""
41    def decorator(func):
42        op_name = operation_name or f"{func.__module__}.{func.__name__}"
43        
44        @functools.wraps(func)
45        def wrapper(*args, **kwargs):
46            with performance_monitor(op_name):
47                return func(*args, **kwargs)
48        return wrapper
49    return decorator
50
51@monitor_performance("Data Processing")
52def process_large_dataset(size=1000000):
53    """Simulate processing a large dataset."""
54    import random
55    
56    # Generate data
57    data = [random.random() for _ in range(size)]
58    
59    # Process data
60    result = {
61        'sum': sum(data),
62        'average': sum(data) / len(data),
63        'max': max(data),
64        'min': min(data)
65    }
66    
67    return result
68
69# Usage
70result = process_large_dataset(500000)
71print(f"Processing result: {result}")

 1import asyncio
 2import aiohttp
 3import time
 4from contextlib import asynccontextmanager
 5
 6class AsyncTimingContext:
 7    """Async context manager for measuring execution time."""
 8    
 9    def __init__(self, description="Async operation"):
10        self.description = description
11        self.start_time = None
12    
13    async def __aenter__(self):
14        self.start_time = time.time()
15        print(f"Starting {self.description}...")
16        return self
17    
18    async def __aexit__(self, exc_type, exc_val, exc_tb):
19        duration = time.time() - self.start_time
20        if exc_type is not None:
21            print(f"{self.description} failed after {duration:.2f}s")
22        else:
23            print(f"{self.description} completed in {duration:.2f}s")
24        return False
25
26@asynccontextmanager
27async def http_session_manager():
28    """Async context manager for HTTP sessions."""
29    print("Creating HTTP session")
30    session = aiohttp.ClientSession()
31    try:
32        yield session
33    finally:
34        await session.close()
35        print("HTTP session closed")
36
37async def fetch_urls(urls):
38    """Fetch multiple URLs concurrently."""
39    async with AsyncTimingContext("Fetching URLs"):
40        async with http_session_manager() as session:
41            tasks = []
42            for url in urls:
43                async def fetch_one(url):
44                    try:
45                        async with session.get(url) as response:
46                            return {
47                                'url': url,
48                                'status': response.status,
49                                'size': len(await response.text())
50                            }
51                    except Exception as e:
52                        return {'url': url, 'error': str(e)}
53                
54                tasks.append(fetch_one(url))
55            
56            results = await asyncio.gather(*tasks)
57            return results
58
59# Usage (would need to be run in an async context)
60async def main():
61    urls = [
62        'https://httpbin.org/delay/1',
63        'https://httpbin.org/json',
64        'https://httpbin.org/html'
65    ]
66    
67    results = await fetch_urls(urls)
68    for result in results:
69        if 'error' in result:
70            print(f"Error fetching {result['url']}: {result['error']}")
71        else:
72            print(f"Fetched {result['url']}: {result['status']} ({result['size']} bytes)")
73
74# To run: asyncio.run(main())