Register a mutation handler for the given mutation types.
(self, callback: Callable, mutation_types: str | Iterable[str] | None = None)
| 575 | ) |
| 576 | |
| 577 | def add_handler(self, callback: Callable, mutation_types: str | Iterable[str] | None = None) -> int: |
| 578 | """Register a mutation handler for the given mutation types.""" |
| 579 | types = self._normalize_types(mutation_types) |
| 580 | |
| 581 | def _dispatch(message: Any) -> None: |
| 582 | if not isinstance(message, dict) or message.get("channel") != self._channel: |
| 583 | return |
| 584 | data = message.get("data") |
| 585 | value = data.get("value") if isinstance(data, dict) else None |
| 586 | if value is None: |
| 587 | return |
| 588 | try: |
| 589 | payload = json.loads(value) |
| 590 | except (ValueError, TypeError): |
| 591 | return |
| 592 | if not isinstance(payload, dict): |
| 593 | return |
| 594 | mutation = dom_mutation_from_payload(payload) |
| 595 | if mutation.type == "attributes" and not mutation.element_id and mutation.element_id != "0": |
| 596 | return |
| 597 | if mutation.type in types: |
| 598 | callback(mutation) |
| 599 | |
| 600 | conn = self._script._conn |
| 601 | with self._lock: |
| 602 | channel_arg = self._channel_argument() |
| 603 | missing = set(types) - self._active_types |
| 604 | if missing: |
| 605 | self._observe_types(channel_arg, missing) |
| 606 | self._active_types |= missing |
| 607 | if not self._handlers: |
| 608 | self._subscription_id = _subscribe_to_event(conn, self.EVENT) |
| 609 | # Register the callback AFTER setup to avoid leaking it if setup fails. |
| 610 | callback_id = conn.add_callback(_EventRef(self.EVENT), _dispatch) |
| 611 | self._handlers[callback_id] = types |
| 612 | return callback_id |
| 613 | |
| 614 | def remove_handler(self, callback_id: int) -> None: |
| 615 | """Remove a handler; tears down observers with the last one.""" |
nothing calls this directly
no test coverage detected