Register edge hooks with selector filtering. Edge objects do not support name filtering. Args: hook_key: Hook stage key. func: Hook callback function. recursion: Reserved for API parity. Recursion is implemented by graphs. ta
(
self,
hook_key: object,
func: Callable,
recursion: bool = False,
target_type: type | tuple[type, ...] | None = None,
target_filter: Callable[[object], bool] | None = None,
selector: Selector | None = None,
)
| 40 | def hooks(self) -> HookManager: |
| 41 | return self._hooks |
| 42 | def hook_register( |
| 43 | self, |
| 44 | hook_key: object, |
| 45 | func: Callable, |
| 46 | recursion: bool = False, |
| 47 | target_type: type | tuple[type, ...] | None = None, |
| 48 | target_filter: Callable[[object], bool] | None = None, |
| 49 | selector: Selector | None = None, |
| 50 | ) -> None: |
| 51 | """ |
| 52 | Register edge hooks with selector filtering. |
| 53 | |
| 54 | Edge objects do not support name filtering. |
| 55 | |
| 56 | Args: |
| 57 | hook_key: Hook stage key. |
| 58 | func: Hook callback function. |
| 59 | recursion: Reserved for API parity. Recursion is implemented by graphs. |
| 60 | target_type: Optional type filter used by the selector. |
| 61 | target_filter: Optional predicate applied to the matched object. |
| 62 | selector: Optional explicit selector. When provided, it overrides `target_type/target_filter`. |
| 63 | """ |
| 64 | predicate = None |
| 65 | if target_filter is not None: |
| 66 | predicate = lambda t: t.obj is not None and target_filter(t.obj) |
| 67 | |
| 68 | selector = build_selector( |
| 69 | selector=selector, |
| 70 | type_filter=target_type, |
| 71 | predicate=predicate, |
| 72 | ) |
| 73 | if selector.match(self): |
| 74 | self._hooks.register(hook_key, func) |
| 75 | |
| 76 | def __str__(self): |
| 77 | return f"Edge from {self._sender.name} to {self._receiver.name} with keys {self._keys}" |