(binding: Any)
| 118 | |
| 119 | |
| 120 | def _resolve_python_binding(binding: Any) -> Callable | None: |
| 121 | if not isinstance(binding, str) or not binding.strip(): |
| 122 | return None |
| 123 | spec = binding.strip() |
| 124 | if spec.startswith("python:"): |
| 125 | spec = spec[len("python:") :] |
| 126 | if ":" in spec: |
| 127 | module_name, attr_name = spec.split(":", 1) |
| 128 | else: |
| 129 | module_name, _, attr_name = spec.rpartition(".") |
| 130 | if not module_name or not attr_name: |
| 131 | raise ValueError(f"Invalid Python binding '{binding}'") |
| 132 | module = importlib.import_module(module_name) |
| 133 | target: Any = module |
| 134 | for part in attr_name.split("."): |
| 135 | target = getattr(target, part) |
| 136 | if not callable(target): |
| 137 | raise TypeError(f"Python binding '{binding}' resolved to non-callable {type(target).__name__}") |
| 138 | return target |
| 139 | |
| 140 | |
| 141 | def _call_predicate(fn: Callable, message: dict[str, object], attributes: dict[str, object]) -> bool: |
no outgoing calls
no test coverage detected