0
# Utilities
1

2
Import hooks, weak references, and utility functions for advanced wrapping scenarios and compatibility. These utilities provide additional functionality for complex use cases and maintain compatibility with different Python versions.
3

4
## Capabilities  
5

6
### Import Hooks
7

8
Import hooks allow you to automatically apply patches or perform actions when specific modules are imported.
9

10
#### register_post_import_hook
11

12
Registers a hook function to be called when a specific module is imported.
13

14
```python { .api }
15
def register_post_import_hook(hook, name):
16
    """
17
    Register hook to be called when module is imported.
18
    
19
    Args:
20
        hook: Callback function or string in format 'module:function'
21
        name: Module name to watch for
22
    """
23
```
24

25
**Usage Example:**
26

27
```python
28
import wrapt
29

30
def patch_requests(module):
31
    """Hook function called when requests module is imported."""
32
    print("Requests module imported, applying patches...")
33
    
34
    def log_wrapper(wrapped, instance, args, kwargs):
35
        print(f"HTTP request to: {args[0] if args else 'unknown'}")
36
        return wrapped(*args, **kwargs)
37
    
38
    wrapt.wrap_function_wrapper(module, 'get', log_wrapper)
39

40
# Register hook before requests is imported
41
wrapt.register_post_import_hook(patch_requests, 'requests')
42

43
# Now when requests is imported, it will be automatically patched
44
import requests  # Triggers the hook
45
response = requests.get('https://httpbin.org/get')  # Will be logged
46
```
47

48
#### when_imported
49

50
Decorator for marking functions as post-import hooks. More convenient than register_post_import_hook for simple cases.
51

52
```python { .api }
53
def when_imported(name):
54
    """
55
    Decorator to register function as post-import hook.
56
    
57
    Args:
58
        name: Module name to watch for
59
        
60
    Returns:
61
        Decorator that registers the function as a hook
62
    """
63
```
64

65
**Usage Example:**
66

67
```python
68
import wrapt
69

70
@wrapt.when_imported('json')
71
def patch_json(module):
72
    """Automatically patch json module when imported."""
73
    print("Patching json module...")
74
    
75
    def timing_wrapper(wrapped, instance, args, kwargs):
76
        import time
77
        start = time.time()
78
        result = wrapped(*args, **kwargs)
79
        print(f"JSON operation took {time.time() - start:.4f}s")
80
        return result
81
    
82
    wrapt.wrap_function_wrapper(module, 'loads', timing_wrapper)
83
    wrapt.wrap_function_wrapper(module, 'dumps', timing_wrapper)
84

85
# Import json - triggers automatic patching
86
import json
87

88
data = json.loads('{"key": "value"}')  # Will show timing
89
result = json.dumps(data)              # Will show timing
90
```
91

92
#### notify_module_loaded
93

94
Manually triggers post-import hooks for a module. Useful for testing or when you need to trigger hooks programmatically.
95

96
```python { .api }
97
def notify_module_loaded(module):
98
    """
99
    Manually trigger post-import hooks for a module.
100
    
101
    Args:
102
        module: Module object that was loaded
103
    """
104
```
105

106
#### discover_post_import_hooks
107

108
Discovers and registers post-import hooks from package entry points. Requires `pkg_resources` to be available.
109

110
```python { .api }
111
def discover_post_import_hooks(group):
112
    """
113
    Discover and register hooks from package entry points.
114
    
115
    Args:
116
        group: Entry point group name to search
117
    """
118
```
119

120
### Weak References
121

122
#### WeakFunctionProxy
123

124
A weak reference proxy for functions that properly handles bound methods, class methods, static methods, and regular functions.
125

126
```python { .api }
127
class WeakFunctionProxy:
128
    def __init__(self, wrapped, callback=None):
129
        """
130
        Create weak reference proxy for function.
131
        
132
        Args:
133
            wrapped: Function to create weak reference for  
134
            callback: Optional callback when reference expires
135
        """
136

137
    @property
138
    def _self_expired(self):
139
        """Boolean flag indicating if reference has expired."""
140

141
    @property
142
    def _self_instance(self):
143
        """Weak reference to instance (for bound methods)."""
144

145
    def __call__(self, *args, **kwargs):
146
        """
147
        Call the weakly referenced function.
148
        
149
        Args:
150
            *args: Positional arguments
151
            **kwargs: Keyword arguments
152
            
153
        Returns:
154
            Result from function call
155
            
156
        Raises:
157
            ReferenceError: If the weak reference has expired
158
        """
159
```
160

161
**Usage Example:**
162

163
```python
164
import wrapt
165
import gc
166

167
class MyClass:
168
    def method(self):
169
        return "method called"
170

171
obj = MyClass()
172

173
# Create weak proxy to bound method
174
weak_method = wrapt.WeakFunctionProxy(obj.method)
175

176
# Call through weak proxy
177
result = weak_method()  # "method called"
178

179
# Delete original object
180
del obj
181
gc.collect()
182

183
# Weak reference is now expired
184
try:
185
    weak_method()
186
except ReferenceError:
187
    print("Weak reference expired")
188
```
189

190
**Callback on Expiration:**
191

192
```python
193
import wrapt
194

195
def cleanup_callback(proxy):
196
    print("Weak reference expired, cleaning up...")
197

198
class MyClass:
199
    def method(self):
200
        return "result"
201

202
obj = MyClass()
203
weak_proxy = wrapt.WeakFunctionProxy(obj.method, cleanup_callback)
204

205
del obj  # Triggers callback
206
```
207

208
### Compatibility Functions
209

210
#### formatargspec
211

212
Forward-compatible implementation of `inspect.formatargspec()` which was removed in Python 3.11.
213

214
```python { .api }
215
def formatargspec(args, varargs=None, varkw=None, defaults=None, 
216
                  kwonlyargs=(), kwonlydefaults={}, annotations={}):
217
    """
218
    Format function argument specification as string.
219
    
220
    Args:
221
        args: List of argument names
222
        varargs: Name of *args parameter (optional)
223
        varkw: Name of **kwargs parameter (optional)  
224
        defaults: Tuple of default values (optional)
225
        kwonlyargs: Tuple of keyword-only argument names (optional)
226
        kwonlydefaults: Dict of keyword-only defaults (optional)
227
        annotations: Dict of type annotations (optional)
228
        
229
    Returns:
230
        String representation of function signature
231
    """
232
```
233

234
**Usage Example:**
235

236
```python
237
import wrapt
238

239
# Format a function signature
240
signature = wrapt.formatargspec(
241
    args=['a', 'b', 'c'],
242
    defaults=[None, 'default'], 
243
    varargs='args',
244
    varkw='kwargs',
245
    annotations={'a': 'int', 'b': 'str', 'return': 'bool'}
246
)
247
print(signature)  # (a: int, b: str = 'default', c=None, *args, **kwargs) -> bool
248
```
249

250
#### getcallargs
251

252
Re-exported from inspect module for backward compatibility. Maps function arguments to their parameter names.
253

254
```python { .api }
255
def getcallargs(func, *positional, **named):
256
    """
257
    Get mapping of function parameters to argument values.
258
    
259
    Args:
260
        func: Function to analyze
261
        *positional: Positional arguments
262
        **named: Named arguments
263
        
264
    Returns:
265
        Dict mapping parameter names to argument values
266
    """
267
```
268

269
**Usage Example:**
270

271
```python
272
import wrapt
273

274
def example_function(a, b=10, *args, **kwargs):
275
    pass
276

277
# Map arguments to parameters
278
mapping = wrapt.getcallargs(example_function, 1, 2, 3, x=4, y=5)
279
print(mapping)  # {'a': 1, 'b': 2, 'args': (3,), 'kwargs': {'x': 4, 'y': 5}}
280
```
281

282
## Advanced Usage
283

284
### Complex Import Hook Scenarios
285

286
Handle complex module import scenarios:
287

288
```python
289
import wrapt
290

291
# Track when multiple related modules are imported
292
modules_to_watch = ['requests', 'urllib3', 'http.client']
293
imported_modules = set()
294

295
def track_http_modules(module):
296
    imported_modules.add(module.__name__)
297
    print(f"HTTP-related module imported: {module.__name__}")
298
    
299
    if len(imported_modules) == len(modules_to_watch):
300
        print("All HTTP modules loaded, applying comprehensive patches...")
301
        apply_comprehensive_http_patches()
302

303
for module_name in modules_to_watch:
304
    wrapt.register_post_import_hook(track_http_modules, module_name)
305

306
def apply_comprehensive_http_patches():
307
    # Apply patches that require multiple modules to be loaded
308
    pass
309
```
310

311
### Weak Reference Cleanup Patterns
312

313
Use weak references for automatic cleanup:
314

315
```python
316
import wrapt
317
import weakref
318

319
class ResourceManager:
320
    def __init__(self):
321
        self.resources = weakref.WeakSet()
322
        self.cleanup_callbacks = []
323
    
324
    def register_resource(self, resource):
325
        self.resources.add(resource)
326
        
327
        # Create weak proxy with cleanup callback
328
        def cleanup(proxy):
329
            print(f"Resource {resource} cleaned up")
330
            
331
        weak_proxy = wrapt.WeakFunctionProxy(
332
            resource.cleanup if hasattr(resource, 'cleanup') else lambda: None,
333
            cleanup
334
        )
335
        self.cleanup_callbacks.append(weak_proxy)
336
    
337
    def cleanup_all(self):
338
        for callback in self.cleanup_callbacks:
339
            try:
340
                callback()
341
            except ReferenceError:
342
                pass  # Already cleaned up
343

344
# Usage
345
manager = ResourceManager()
346

347
class Resource:
348
    def cleanup(self):
349
        print("Resource cleanup called")
350

351
resource = Resource()
352
manager.register_resource(resource)
353

354
del resource  # Triggers automatic cleanup
355
```
356

357
### Custom Hook Discovery
358

359
Implement custom hook discovery mechanisms:
360

361
```python
362
import wrapt
363
import importlib
364
import pkgutil
365

366
def discover_plugin_hooks(plugin_namespace):
367
    """Discover hooks from plugin packages."""
368
    try:
369
        namespace_pkg = importlib.import_module(plugin_namespace)
370
        for finder, name, ispkg in pkgutil.iter_modules(
371
            namespace_pkg.__path__, 
372
            namespace_pkg.__name__ + "."
373
        ):
374
            try:
375
                plugin_module = importlib.import_module(name)
376
                if hasattr(plugin_module, 'register_hooks'):
377
                    plugin_module.register_hooks()
378
                    print(f"Registered hooks from {name}")
379
            except ImportError as e:
380
                print(f"Failed to load plugin {name}: {e}")
381
    except ImportError:
382
        print(f"Plugin namespace {plugin_namespace} not found")
383

384
# Example plugin structure:
385
# myapp_plugins/
386
#   __init__.py
387
#   database_plugin.py  # Contains register_hooks() function
388
#   cache_plugin.py     # Contains register_hooks() function
389

390
discover_plugin_hooks('myapp_plugins')
391
```
392

393
## Performance Considerations
394

395
### Efficient Hook Management
396

397
For applications with many hooks, consider efficient management:
398

399
```python
400
import wrapt
401
from collections import defaultdict
402

403
class HookManager:
404
    def __init__(self):
405
        self.hooks_by_module = defaultdict(list)
406
        self.registered_modules = set()
407
    
408
    def register_hook(self, module_name, hook_func):
409
        """Register hook with batching for efficiency."""
410
        self.hooks_by_module[module_name].append(hook_func)
411
        
412
        if module_name not in self.registered_modules:
413
            wrapt.register_post_import_hook(
414
                self._execute_hooks_for_module, 
415
                module_name
416
            )
417
            self.registered_modules.add(module_name)
418
    
419
    def _execute_hooks_for_module(self, module):
420
        """Execute all hooks for a module efficiently."""
421
        hooks = self.hooks_by_module[module.__name__]
422
        for hook in hooks:
423
            try:
424
                hook(module)
425
            except Exception as e:
426
                print(f"Hook failed for {module.__name__}: {e}")
427

428
# Global hook manager
429
hook_manager = HookManager()
430

431
# Register multiple hooks efficiently
432
hook_manager.register_hook('requests', lambda m: print("Hook 1"))
433
hook_manager.register_hook('requests', lambda m: print("Hook 2")) 
434
hook_manager.register_hook('requests', lambda m: print("Hook 3"))
435

436
import requests  # All hooks execute together
437
```