0
# Patching and Monkey Patching
1
2
Comprehensive utilities for applying patches, wrapping objects, and monkey patching with proper cleanup and context management. These functions provide safe and robust ways to modify existing code at runtime while preserving original behavior and enabling easy rollback.
3
4
## Capabilities
5
6
### Object Wrapping
7
8
#### wrap_function_wrapper
9
10
Convenience function to wrap a function with FunctionWrapper. This is the most commonly used patching function.
11
12
```python { .api }
13
def wrap_function_wrapper(module, name, wrapper):
14
"""
15
Wrap a function with FunctionWrapper.
16
17
Args:
18
module: Module object or module name string
19
name: Dotted path to function (e.g., 'ClassName.method_name')
20
wrapper: Wrapper function with signature (wrapped, instance, args, kwargs) -> result
21
22
Returns:
23
The FunctionWrapper instance that was applied
24
"""
25
```
26
27
**Usage Example:**
28
29
```python
30
import wrapt
31
import time
32
33
def timing_wrapper(wrapped, instance, args, kwargs):
34
start = time.time()
35
result = wrapped(*args, **kwargs)
36
end = time.time()
37
print(f"{wrapped.__name__} took {end - start:.4f} seconds")
38
return result
39
40
# Patch the time.sleep function
41
wrapt.wrap_function_wrapper('time', 'sleep', timing_wrapper)
42
43
# Now all calls to time.sleep will be timed
44
time.sleep(0.1) # Prints timing information
45
```
46
47
#### wrap_object
48
49
Wraps an object attribute with a factory function. More flexible than wrap_function_wrapper as it supports any factory function.
50
51
```python { .api }
52
def wrap_object(module, name, factory, args=(), kwargs={}):
53
"""
54
Wrap an object attribute with a factory function.
55
56
Args:
57
module: Module object or module name string
58
name: Dotted path to attribute
59
factory: Factory function to create wrapper
60
args: Arguments for factory (optional)
61
kwargs: Keyword arguments for factory (optional)
62
63
Returns:
64
The wrapper created by the factory
65
"""
66
```
67
68
**Usage Example:**
69
70
```python
71
import wrapt
72
73
def proxy_factory(wrapped, *args, **kwargs):
74
class LoggingProxy(wrapt.ObjectProxy):
75
def __call__(self, *call_args, **call_kwargs):
76
print(f"Calling {wrapped.__name__}")
77
return super().__call__(*call_args, **call_kwargs)
78
return LoggingProxy(wrapped)
79
80
# Wrap a function with custom proxy
81
def my_function():
82
return "Hello"
83
84
wrapt.wrap_object(__name__, 'my_function', proxy_factory)
85
result = my_function() # Logs the call
86
```
87
88
#### wrap_object_attribute
89
90
Wraps instance attributes using a descriptor-based approach. Useful for wrapping attributes that are set on instances rather than classes.
91
92
```python { .api }
93
def wrap_object_attribute(module, name, factory, args=(), kwargs={}):
94
"""
95
Wrap instance attributes using descriptor approach.
96
97
Args:
98
module: Module object or module name string
99
name: Dotted path to attribute
100
factory: Factory function to create wrapper
101
args: Arguments for factory (optional)
102
kwargs: Keyword arguments for factory (optional)
103
104
Returns:
105
The AttributeWrapper descriptor
106
"""
107
```
108
109
### Patch Resolution
110
111
#### resolve_path
112
113
Resolves a dotted path to get the parent object, attribute name, and original attribute value.
114
115
```python { .api }
116
def resolve_path(module, name):
117
"""
118
Resolve dotted path to parent object and attribute.
119
120
Args:
121
module: Module object or module name string
122
name: Dotted attribute path (e.g., 'ClassName.method_name')
123
124
Returns:
125
tuple: (parent_object, attribute_name, original_value)
126
"""
127
```
128
129
**Usage Example:**
130
131
```python
132
import wrapt
133
import os
134
135
# Resolve path to os.path.exists
136
parent, attr_name, original = wrapt.resolve_path('os', 'path.exists')
137
print(f"Parent: {parent}") # <module 'os.path'>
138
print(f"Attribute: {attr_name}") # 'exists'
139
print(f"Original: {original}") # <function exists at ...>
140
```
141
142
#### apply_patch
143
144
Low-level function to apply a replacement to an attribute of an object.
145
146
```python { .api }
147
def apply_patch(parent, attribute, replacement):
148
"""
149
Apply replacement to an attribute of an object.
150
151
Args:
152
parent: Parent object
153
attribute: Attribute name string
154
replacement: New value to set
155
"""
156
```
157
158
### Decorator Factories
159
160
#### function_wrapper
161
162
Creates a decorator that converts a function into a FunctionWrapper-compatible wrapper format.
163
164
```python { .api }
165
def function_wrapper(wrapper):
166
"""
167
Convert function to FunctionWrapper-compatible format.
168
169
Args:
170
wrapper: Function to convert to wrapper format
171
172
Returns:
173
Decorator function that applies the wrapper
174
"""
175
```
176
177
**Usage Example:**
178
179
```python
180
import wrapt
181
182
@wrapt.function_wrapper
183
def my_wrapper(wrapped, instance, args, kwargs):
184
print(f"Wrapped function: {wrapped.__name__}")
185
return wrapped(*args, **kwargs)
186
187
@my_wrapper
188
def my_function():
189
return "Hello"
190
191
my_function() # Prints: "Wrapped function: my_function"
192
```
193
194
#### patch_function_wrapper
195
196
Decorator factory for patching functions. Creates a decorator that patches the specified function when applied.
197
198
```python { .api }
199
def patch_function_wrapper(module, name, enabled=None):
200
"""
201
Create decorator that patches specified function.
202
203
Args:
204
module: Module object or module name string
205
name: Dotted path to function
206
enabled: Enable/disable flag or callable (optional)
207
208
Returns:
209
Decorator function that patches the target
210
"""
211
```
212
213
**Usage Example:**
214
215
```python
216
import wrapt
217
import requests
218
219
@wrapt.patch_function_wrapper('requests', 'get')
220
def log_requests(wrapped, instance, args, kwargs):
221
print(f"Making request to: {args[0] if args else 'unknown'}")
222
return wrapped(*args, **kwargs)
223
224
# Now all requests.get calls will be logged
225
response = requests.get('https://httpbin.org/get')
226
```
227
228
#### transient_function_wrapper
229
230
Creates a decorator that temporarily patches a function only during the execution of the decorated function.
231
232
```python { .api }
233
def transient_function_wrapper(module, name):
234
"""
235
Create decorator for temporary function patching.
236
237
Args:
238
module: Module object or module name string
239
name: Dotted path to function
240
241
Returns:
242
Decorator function that temporarily patches target
243
"""
244
```
245
246
**Usage Example:**
247
248
```python
249
import wrapt
250
import time
251
252
@wrapt.transient_function_wrapper('time', 'sleep')
253
def no_sleep_wrapper(wrapped, instance, args, kwargs):
254
print(f"Skipping sleep({args[0] if args else 0})")
255
# Don't call the original function - skip sleeping
256
return None
257
258
def normal_function():
259
print("Before sleep")
260
time.sleep(1) # This will actually sleep
261
print("After sleep")
262
263
@no_sleep_wrapper
264
def fast_function():
265
print("Before sleep")
266
time.sleep(1) # This will be skipped
267
print("After sleep")
268
269
normal_function() # Takes 1 second
270
fast_function() # Instant, sleep is patched only during execution
271
```
272
273
## Advanced Usage
274
275
### Conditional Patching
276
277
Use enabled parameter to control when patches are active:
278
279
```python
280
import wrapt
281
import os
282
283
def debug_wrapper(wrapped, instance, args, kwargs):
284
print(f"DEBUG: {wrapped.__name__} called")
285
return wrapped(*args, **kwargs)
286
287
# Only patch when DEBUG environment variable is set
288
debug_enabled = lambda: os.environ.get('DEBUG') == '1'
289
290
wrapt.wrap_function_wrapper(
291
'builtins', 'print', debug_wrapper,
292
enabled=debug_enabled
293
)
294
```
295
296
### Context-Aware Patching
297
298
Create patches that are aware of their execution context:
299
300
```python
301
import wrapt
302
import threading
303
304
def thread_aware_wrapper(wrapped, instance, args, kwargs):
305
thread_id = threading.current_thread().ident
306
print(f"Thread {thread_id}: {wrapped.__name__}")
307
return wrapped(*args, **kwargs)
308
309
# Patch will show which thread is executing
310
wrapt.wrap_function_wrapper('time', 'sleep', thread_aware_wrapper)
311
312
import concurrent.futures
313
import time
314
315
def worker(n):
316
time.sleep(0.1) # Will show thread ID
317
return n * 2
318
319
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
320
futures = [executor.submit(worker, i) for i in range(5)]
321
results = [f.result() for f in futures]
322
```
323
324
### Patch Cleanup and Management
325
326
For production code, consider implementing patch management:
327
328
```python
329
import wrapt
330
import atexit
331
332
class PatchManager:
333
def __init__(self):
334
self.patches = []
335
336
def patch_function(self, module, name, wrapper, enabled=None):
337
"""Apply patch and track it for cleanup."""
338
original_wrapper = wrapt.wrap_function_wrapper(
339
module, name, wrapper, enabled=enabled
340
)
341
self.patches.append((module, name, original_wrapper))
342
return original_wrapper
343
344
def remove_all_patches(self):
345
"""Remove all tracked patches."""
346
for module, name, wrapper in self.patches:
347
# Note: wrapt doesn't provide built-in patch removal
348
# You would need to implement this based on your needs
349
pass
350
self.patches.clear()
351
352
# Global patch manager
353
patch_manager = PatchManager()
354
atexit.register(patch_manager.remove_all_patches)
355
356
# Use managed patching
357
def my_wrapper(wrapped, instance, args, kwargs):
358
return wrapped(*args, **kwargs)
359
360
patch_manager.patch_function('time', 'sleep', my_wrapper)
361
```
362
363
## Error Handling
364
365
Patching operations can fail for various reasons. Handle errors appropriately:
366
367
```python
368
import wrapt
369
370
def safe_patch_function(module, name, wrapper):
371
"""Safely patch function with error handling."""
372
try:
373
return wrapt.wrap_function_wrapper(module, name, wrapper)
374
except AttributeError as e:
375
print(f"Failed to patch {module}.{name}: {e}")
376
return None
377
except ImportError as e:
378
print(f"Module {module} not available: {e}")
379
return None
380
381
# Safe patching
382
wrapper = safe_patch_function('nonexistent_module', 'function', lambda w,i,a,k: w(*a,**k))
383
if wrapper:
384
print("Patch applied successfully")
385
else:
386
print("Patch failed, continuing without it")
387
```
388
389
## Best Practices
390
391
1. **Use descriptive wrapper names** for debugging
392
2. **Always preserve original behavior** unless intentionally changing it
393
3. **Handle exceptions properly** to avoid breaking wrapped code
394
4. **Consider thread safety** when patching shared resources
395
5. **Document patches clearly** for maintenance
396
6. **Test patches thoroughly** including edge cases and error conditions