Module for decorators, wrappers and monkey patching.
Universal decorator factory and synchronization utilities for creating robust decorators with proper signature preservation. These tools enable the creation of decorators that work correctly across functions, methods, classes, and other callable objects while preserving introspection capabilities.
The main decorator factory for creating robust decorators with proper signature preservation. This is the recommended way to create decorators in wrapt.
def decorator(wrapper=None, enabled=None, adapter=None, proxy=FunctionWrapper):
"""
Universal decorator factory for creating robust decorators.
Args:
wrapper: Wrapper function with signature (wrapped, instance, args, kwargs) -> result
enabled: Boolean or callable to enable/disable decorator (optional)
adapter: Function or signature spec for signature adaptation (optional)
proxy: Proxy class to use, defaults to FunctionWrapper (optional)
Returns:
Decorator function or partial decorator if wrapper not provided
"""Basic Usage:
import wrapt
@wrapt.decorator
def my_decorator(wrapped, instance, args, kwargs):
print(f"Calling {wrapped.__name__}")
result = wrapped(*args, **kwargs)
print(f"Result: {result}")
return result
@my_decorator
def greet(name):
return f"Hello, {name}!"
result = greet("Alice") # Logs call and resultPartial Application:
import wrapt
# Create reusable decorator factory
def timing_decorator(enabled=True):
@wrapt.decorator(enabled=enabled)
def wrapper(wrapped, instance, args, kwargs):
import time
start = time.time()
result = wrapped(*args, **kwargs)
end = time.time()
print(f"{wrapped.__name__} took {end - start:.4f}s")
return result
return wrapper
# Use with different configurations
@timing_decorator(enabled=True)
def slow_function():
import time
time.sleep(0.1)
return "done"
@timing_decorator(enabled=False)
def fast_function():
return "instant"
slow_function() # Shows timing
fast_function() # No timing outputBase class for creating adapter factories that modify function signatures or behavior.
class AdapterFactory:
"""
Base class for adapter factories.
"""
def __call__(self, wrapped):
"""
Abstract method to be implemented by subclasses.
Args:
wrapped: The function being adapted
Returns:
Adapter function or signature specification
"""Alias for DelegatedAdapterFactory class - a factory that delegates to another factory function.
adapter_factory = DelegatedAdapterFactory
# Usage: adapter_factory(factory_function)Usage Example:
import wrapt
import inspect
def signature_adapter_factory(wrapped):
"""Factory that creates signature adapter."""
sig = inspect.signature(wrapped)
def adapter(*args, **kwargs):
# Validate arguments against signature
bound = sig.bind(*args, **kwargs)
bound.apply_defaults()
return wrapped(**bound.arguments)
return adapter
@wrapt.decorator(adapter=wrapt.adapter_factory(signature_adapter_factory))
def validating_decorator(wrapped, instance, args, kwargs):
print(f"Validated call to {wrapped.__name__}")
return wrapped(*args, **kwargs)
@validating_decorator
def add_numbers(a: int, b: int = 10) -> int:
return a + b
result = add_numbers(5) # Uses default b=10, validatedDecorator and context manager for thread synchronization using locks. Provides automatic thread safety for functions and methods.
def synchronized(wrapped):
"""
Decorator/context manager for thread synchronization.
Args:
wrapped: Function to synchronize OR lock object with acquire()/release()
Returns:
Synchronized decorator or context manager
"""As Decorator:
import wrapt
import threading
import time
class Counter:
def __init__(self):
self.value = 0
@wrapt.synchronized
def increment(self):
# This method is thread-safe
temp = self.value
time.sleep(0.001) # Simulate work
self.value = temp + 1
@wrapt.synchronized
def get_value(self):
return self.value
counter = Counter()
def worker():
for _ in range(100):
counter.increment()
threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
print(counter.get_value()) # Should be 1000, not less due to race conditionsAs Context Manager:
import wrapt
import threading
# Using custom lock
my_lock = threading.RLock()
@wrapt.synchronized(my_lock)
def critical_section():
print("In critical section")
time.sleep(0.1)
# Or as context manager
with wrapt.synchronized(my_lock):
print("Also in critical section")
time.sleep(0.1)Create decorators that are conditionally applied:
import wrapt
import os
def debug_decorator(enabled=None):
if enabled is None:
enabled = lambda: os.environ.get('DEBUG') == '1'
@wrapt.decorator(enabled=enabled)
def wrapper(wrapped, instance, args, kwargs):
print(f"DEBUG: {wrapped.__name__} called with {args}, {kwargs}")
result = wrapped(*args, **kwargs)
print(f"DEBUG: {wrapped.__name__} returned {result}")
return result
return wrapper
@debug_decorator()
def my_function(x, y):
return x + y
# Only logs when DEBUG=1 environment variable is set
result = my_function(2, 3)The decorator function works with classes as well as functions:
import wrapt
@wrapt.decorator
def class_wrapper(wrapped, instance, args, kwargs):
print(f"Creating instance of {wrapped.__name__}")
instance = wrapped(*args, **kwargs)
print(f"Created: {instance}")
return instance
@class_wrapper
class MyClass:
def __init__(self, name):
self.name = name
def __repr__(self):
return f"MyClass({self.name})"
obj = MyClass("test") # Logs creationDecorators that behave differently for functions vs methods:
import wrapt
@wrapt.decorator
def smart_decorator(wrapped, instance, args, kwargs):
if instance is None:
print(f"Function call: {wrapped.__name__}")
else:
print(f"Method call: {wrapped.__name__} on {type(instance).__name__}")
return wrapped(*args, **kwargs)
@smart_decorator
def standalone_function():
return "function result"
class MyClass:
@smart_decorator
def method(self):
return "method result"
standalone_function() # Logs: "Function call: standalone_function"
obj = MyClass()
obj.method() # Logs: "Method call: method on MyClass"Multiple decorators work correctly with wrapt:
import wrapt
@wrapt.decorator
def decorator1(wrapped, instance, args, kwargs):
print("Decorator 1 - before")
result = wrapped(*args, **kwargs)
print("Decorator 1 - after")
return result
@wrapt.decorator
def decorator2(wrapped, instance, args, kwargs):
print("Decorator 2 - before")
result = wrapped(*args, **kwargs)
print("Decorator 2 - after")
return result
@decorator1
@decorator2
def my_function():
print("Function executing")
return "result"
my_function()
# Output:
# Decorator 1 - before
# Decorator 2 - before
# Function executing
# Decorator 2 - after
# Decorator 1 - afterUse custom proxy classes with the decorator:
import wrapt
class CustomProxy(wrapt.FunctionWrapper):
def __init__(self, wrapped, wrapper):
super().__init__(wrapped, wrapper)
self._self_call_count = 0
def __call__(self, *args, **kwargs):
self._self_call_count += 1
print(f"Call #{self._self_call_count}")
return super().__call__(*args, **kwargs)
@wrapt.decorator(proxy=CustomProxy)
def counting_decorator(wrapped, instance, args, kwargs):
return wrapped(*args, **kwargs)
@counting_decorator
def my_function():
return "result"
my_function() # Call #1
my_function() # Call #2The decorator function preserves function signatures for introspection:
import wrapt
import inspect
@wrapt.decorator
def preserving_decorator(wrapped, instance, args, kwargs):
return wrapped(*args, **kwargs)
@preserving_decorator
def example_function(a: int, b: str = "default") -> str:
"""Example function with type hints."""
return f"{a}: {b}"
# Signature is preserved
sig = inspect.signature(example_function)
print(sig) # (a: int, b: str = 'default') -> str
# Docstring is preserved
print(example_function.__doc__) # "Example function with type hints."
# Name is preserved
print(example_function.__name__) # "example_function"Decorators created with wrapt properly preserve exception information:
import wrapt
@wrapt.decorator
def error_handling_decorator(wrapped, instance, args, kwargs):
try:
return wrapped(*args, **kwargs)
except Exception as e:
print(f"Exception in {wrapped.__name__}: {e}")
raise # Re-raise preserving traceback
@error_handling_decorator
def failing_function():
raise ValueError("Something went wrong")
try:
failing_function()
except ValueError:
import traceback
traceback.print_exc() # Shows original tracebackInstall with Tessl CLI
npx tessl i tessl/pypi-wrapt