Patching and Monkey Patching

def wrap_function_wrapper(module, name, wrapper):
    """
    Wrap a function with FunctionWrapper.
    
    Args:
        module: Module object or module name string  
        name: Dotted path to function (e.g., 'ClassName.method_name')
        wrapper: Wrapper function with signature (wrapped, instance, args, kwargs) -> result
        
    Returns:
        The FunctionWrapper instance that was applied
    """

import wrapt
import time

def timing_wrapper(wrapped, instance, args, kwargs):
    start = time.time()
    result = wrapped(*args, **kwargs)
    end = time.time()
    print(f"{wrapped.__name__} took {end - start:.4f} seconds")
    return result

# Patch the time.sleep function
wrapt.wrap_function_wrapper('time', 'sleep', timing_wrapper)

# Now all calls to time.sleep will be timed
time.sleep(0.1)  # Prints timing information

def wrap_object(module, name, factory, args=(), kwargs={}):
    """
    Wrap an object attribute with a factory function.
    
    Args:
        module: Module object or module name string
        name: Dotted path to attribute
        factory: Factory function to create wrapper
        args: Arguments for factory (optional)
        kwargs: Keyword arguments for factory (optional)
        
    Returns:
        The wrapper created by the factory
    """

import wrapt

def proxy_factory(wrapped, *args, **kwargs):
    class LoggingProxy(wrapt.ObjectProxy):
        def __call__(self, *call_args, **call_kwargs):
            print(f"Calling {wrapped.__name__}")
            return super().__call__(*call_args, **call_kwargs)
    return LoggingProxy(wrapped)

# Wrap a function with custom proxy  
def my_function():
    return "Hello"

wrapt.wrap_object(__name__, 'my_function', proxy_factory)
result = my_function()  # Logs the call

def wrap_object_attribute(module, name, factory, args=(), kwargs={}):
    """
    Wrap instance attributes using descriptor approach.
    
    Args:
        module: Module object or module name string
        name: Dotted path to attribute  
        factory: Factory function to create wrapper
        args: Arguments for factory (optional)
        kwargs: Keyword arguments for factory (optional)
        
    Returns:
        The AttributeWrapper descriptor
    """

def resolve_path(module, name):
    """
    Resolve dotted path to parent object and attribute.
    
    Args:
        module: Module object or module name string
        name: Dotted attribute path (e.g., 'ClassName.method_name')
        
    Returns:
        tuple: (parent_object, attribute_name, original_value)
    """

import wrapt
import os

# Resolve path to os.path.exists
parent, attr_name, original = wrapt.resolve_path('os', 'path.exists')
print(f"Parent: {parent}")           # <module 'os.path'>
print(f"Attribute: {attr_name}")     # 'exists'  
print(f"Original: {original}")       # <function exists at ...>

def apply_patch(parent, attribute, replacement):
    """
    Apply replacement to an attribute of an object.
    
    Args:
        parent: Parent object
        attribute: Attribute name string
        replacement: New value to set
    """

def function_wrapper(wrapper):
    """
    Convert function to FunctionWrapper-compatible format.
    
    Args:
        wrapper: Function to convert to wrapper format
        
    Returns:
        Decorator function that applies the wrapper
    """

import wrapt

@wrapt.function_wrapper
def my_wrapper(wrapped, instance, args, kwargs):
    print(f"Wrapped function: {wrapped.__name__}")
    return wrapped(*args, **kwargs)

@my_wrapper  
def my_function():
    return "Hello"

my_function()  # Prints: "Wrapped function: my_function"

def patch_function_wrapper(module, name, enabled=None):
    """
    Create decorator that patches specified function.
    
    Args:
        module: Module object or module name string
        name: Dotted path to function
        enabled: Enable/disable flag or callable (optional)
        
    Returns:
        Decorator function that patches the target
    """

import wrapt
import requests

@wrapt.patch_function_wrapper('requests', 'get')
def log_requests(wrapped, instance, args, kwargs):
    print(f"Making request to: {args[0] if args else 'unknown'}")
    return wrapped(*args, **kwargs)

# Now all requests.get calls will be logged
response = requests.get('https://httpbin.org/get')

def transient_function_wrapper(module, name):
    """
    Create decorator for temporary function patching.
    
    Args:
        module: Module object or module name string  
        name: Dotted path to function
        
    Returns:
        Decorator function that temporarily patches target
    """

import wrapt
import time

@wrapt.transient_function_wrapper('time', 'sleep')
def no_sleep_wrapper(wrapped, instance, args, kwargs):
    print(f"Skipping sleep({args[0] if args else 0})")
    # Don't call the original function - skip sleeping
    return None

def normal_function():
    print("Before sleep")
    time.sleep(1)  # This will actually sleep
    print("After sleep")

@no_sleep_wrapper
def fast_function():
    print("Before sleep")  
    time.sleep(1)  # This will be skipped
    print("After sleep")

normal_function()  # Takes 1 second
fast_function()    # Instant, sleep is patched only during execution

import wrapt
import os

def debug_wrapper(wrapped, instance, args, kwargs):
    print(f"DEBUG: {wrapped.__name__} called")
    return wrapped(*args, **kwargs)

# Only patch when DEBUG environment variable is set
debug_enabled = lambda: os.environ.get('DEBUG') == '1'

wrapt.wrap_function_wrapper(
    'builtins', 'print', debug_wrapper, 
    enabled=debug_enabled
)

import wrapt
import threading

def thread_aware_wrapper(wrapped, instance, args, kwargs):
    thread_id = threading.current_thread().ident
    print(f"Thread {thread_id}: {wrapped.__name__}")
    return wrapped(*args, **kwargs)

# Patch will show which thread is executing
wrapt.wrap_function_wrapper('time', 'sleep', thread_aware_wrapper)

import concurrent.futures
import time

def worker(n):
    time.sleep(0.1)  # Will show thread ID
    return n * 2

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(worker, i) for i in range(5)]
    results = [f.result() for f in futures]

import wrapt
import atexit

class PatchManager:
    def __init__(self):
        self.patches = []
    
    def patch_function(self, module, name, wrapper, enabled=None):
        """Apply patch and track it for cleanup."""
        original_wrapper = wrapt.wrap_function_wrapper(
            module, name, wrapper, enabled=enabled
        )
        self.patches.append((module, name, original_wrapper))
        return original_wrapper
    
    def remove_all_patches(self):
        """Remove all tracked patches."""
        for module, name, wrapper in self.patches:
            # Note: wrapt doesn't provide built-in patch removal
            # You would need to implement this based on your needs
            pass
        self.patches.clear()

# Global patch manager
patch_manager = PatchManager()
atexit.register(patch_manager.remove_all_patches)

# Use managed patching
def my_wrapper(wrapped, instance, args, kwargs):
    return wrapped(*args, **kwargs)

patch_manager.patch_function('time', 'sleep', my_wrapper)

import wrapt

def safe_patch_function(module, name, wrapper):
    """Safely patch function with error handling."""
    try:
        return wrapt.wrap_function_wrapper(module, name, wrapper)
    except AttributeError as e:
        print(f"Failed to patch {module}.{name}: {e}")
        return None
    except ImportError as e:
        print(f"Module {module} not available: {e}")
        return None

# Safe patching
wrapper = safe_patch_function('nonexistent_module', 'function', lambda w,i,a,k: w(*a,**k))
if wrapper:
    print("Patch applied successfully")
else:
    print("Patch failed, continuing without it")