ddp_utils.browser.observers.mutation

import ddp_utils.browser.observers.mutation

Control a browser-side DOM MutationObserver through the unified transport.

Examples

Start, read, and stop an observer:

from ddp_utils.browser.observers import MutationObserver

observer = MutationObserver(browser, selector="#results")
observer.start()
records = observer.read()
observer.stop()

class ddp_utils.browser.observers.mutation.MutationEvent(type: str, target: str, timestamp: float, attribute_name: str | None = None, old_value: str | None = None, new_value: str | None = None, added_nodes: Tuple[str, ...] = (), removed_nodes: Tuple[str, ...] = ())

Bases: object

Serializable subset of one browser MutationRecord.

Variables

Name	Type	Description
type	str	Browser mutation type, such as attributes or childList.
target	str	Compact description of the mutated DOM target.
timestamp	float	Browser-side event timestamp.
attribute_name	str | None	Changed attribute name, or None for other events.
old_value	str | None	Previous attribute or character-data value when requested.
new_value	str | None	Current attribute or character-data value when available.
added_nodes	Tuple[str, ...]	Descriptions of nodes added by a child-list mutation.
removed_nodes	Tuple[str, ...]	Descriptions of nodes removed by a child-list mutation.

Examples

Represent one attribute mutation:

event = MutationEvent(
    type="attributes",
    target="button#submit",
    timestamp=1.0,
    attribute_name="disabled",
)

classmethod from_payload(payload: Dict[str, Any]) → MutationEvent

Build a typed event from WebDriver’s JSON-compatible payload.

Parameters

Name	Type	Description
payload	Dict[str, Any]	Browser event mapping. Missing scalar fields receive empty or zero defaults, and missing node collections become empty tuples.

Returns

Type	Description
MutationEvent	A normalized immutable event whose node descriptions are strings.

Raises

Exception	Description
TypeError	A supplied field cannot be iterated or converted to its required representation.
ValueError	timestamp cannot be converted to float.

Examples

Convert a browser payload:

event = MutationEvent.from_payload(
    {"type": "childList", "target": "main", "timestamp": 1}
)

class ddp_utils.browser.observers.mutation.MutationObserver(driver: Any, *, selector: str | None = None, child_list: bool = True, attributes: bool = True, character_data: bool = True, subtree: bool = True, attribute_filter: Iterable[str] | None = None, include_old_value: bool = True, max_records: int = 1000, poll_interval: float = 0.1, timeout: float = 30.0)

Bases: object

Observe DOM mutations in-page and consume them through Python.

JavaScript owns mutation collection, so changes between Python polling cycles are not lost. Python performs no background WebDriver calls. The in-page buffer is bounded by max_records to prevent unbounded memory growth. A full page navigation destroys the browser-side observer; call start() again after navigation.

Examples

Use this public operation:

instance = MutationObserver(driver)

Configure a browser-side DOM observer without starting it yet.

Parameters

Name	Type	Description
driver	Any	WebDriver used to run the in-page observer script.
selector	Optional[str]	CSS selector of the observed element; the document element when omitted.
child_list	bool	Report added and removed child nodes.
attributes	bool	Report attribute changes.
character_data	bool	Report text node changes.
subtree	bool	Observe descendants of the target as well.
attribute_filter	Optional[Iterable[str]]	Attribute names to report; requires attributes=True.
include_old_value	bool	Include previous values of changed attributes and text.
max_records	int	Maximum number of records kept in the in-page buffer; must be greater than zero.
poll_interval	float	Seconds between polls; must be greater than zero.
timeout	float	Default time limit for waiting operations, in seconds; cannot be negative.

Raises

Exception	Description
ValueError	If no mutation type is enabled, max_records or poll_interval is not greater than zero, timeout is negative, or attribute_filter is used without attributes=True.

start() → MutationObserver

Inject and start the browser MutationObserver.

Starting again replaces the browser-side state associated with this Python instance and resets its last known dropped-record count.

Returns

Type	Description
MutationObserver	This active observer for fluent use and context-manager entry.

Raises

Exception	Description
MutationObserverTargetError	The configured observation root does not exist in the current document.
MutationObserverError	The browser rejects script execution.

Examples

Start explicitly and always release browser-side state:

observer.start()
try:
    events = observer.wait_for_change(timeout=1.0)
finally:
    observer.stop()

stop() → Tuple[MutationEvent, ...]

Disconnect the observer and return its remaining buffered events.

Returns

Type	Description
Tuple[MutationEvent, …]	Remaining events in collection order. An empty tuple is a soft no-op when this instance is already inactive.

Raises

Exception	Description
MutationObserverError	The browser rejects the stop command.

Examples

Preserve final events during cleanup:

remaining = observer.stop()

take_records() → Tuple[MutationEvent, ...]

Atomically consume every currently buffered mutation event.

Returns

Type	Description
Tuple[MutationEvent, …]	Buffered events in collection order. An empty tuple means the observer is inactive, its page state disappeared, or no events were available.

Raises

Exception	Description
MutationObserverError	The browser rejects the read command.

Examples

Drain the current browser-side buffer:

events = observer.take_records()

wait_for_change(timeout: float | None = None, *, minimum: int = 1) → Tuple[MutationEvent, ...]

Wait for buffered mutations and return them, or an empty tuple.

Parameters

Name	Type	Description
timeout	float | None	Maximum seconds to wait. None uses the observer’s configured default; zero performs one immediate check.
minimum	int	Positive number of buffered events required before the entire buffer is consumed and returned.

Returns

Type	Description
Tuple[MutationEvent, …]	All buffered events once minimum is reached. An empty tuple indicates timeout or inactive browser-side state.

Raises

Exception	Description
ValueError	minimum is not positive or the resolved timeout is negative.
MutationObserverError	Browser-side state inspection fails.

Examples

Wait briefly for at least two mutations:

events = observer.wait_for_change(timeout=1.0, minimum=2)

wait_for_appearance(selector: str, timeout: float | None = None) → Any | None

Wait for an element to appear relative to the observation root.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider element reference when found, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for a dynamically inserted dialog:

dialog = observer.wait_for_appearance("[role='dialog']", timeout=5)

wait_for_disappearance(selector: str, timeout: float | None = None) → bool

Wait until an element relative to the observation root is absent.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
bool	True when the element becomes absent; False on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for a loading indicator to disappear:

disappeared = observer.wait_for_disappearance(".loading", timeout=5)

wait_for_attribute(selector: str, attribute: str, timeout: float | None = None) → Any | None

Wait until an element exists and has the requested attribute.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
attribute	str	Attribute name whose presence is required.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider element reference when the attribute exists, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for accessibility state to be published:

element = observer.wait_for_attribute(
    "#status", "aria-live", timeout=5
)

wait_for_attribute_value(selector: str, attribute: str, expected: Any, timeout: float | None = None) → Any | None

Wait until an attribute equals the expected JSON-compatible value.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
attribute	str	Attribute name to inspect.
expected	Any	Exact JSON-compatible value expected from the browser.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider element reference when the value matches, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for a widget to report its expanded state:

element = observer.wait_for_attribute_value(
    "#menu", "aria-expanded", "true", timeout=5
)

wait_for_attribute_removal(selector: str, attribute: str, timeout: float | None = None) → bool

Wait until an existing element no longer has an attribute.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
attribute	str	Attribute name that must disappear.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
bool	True when the element exists without the attribute; False on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait until a control is no longer marked busy:

ready = observer.wait_for_attribute_removal(
    "#results", "aria-busy", timeout=5
)

wait_for_text(selector: str, text: str, timeout: float | None = None, *, exact: bool = False, case_sensitive: bool = True) → Any | None

Wait for exact or contained text on an element under the root.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
text	str	Text to match exactly or as a substring.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.
exact	bool	Require the complete element text to equal text.
case_sensitive	bool	Preserve case during comparison when True.

Returns

Type	Description
Any | None	Provider element reference when its text matches, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for a case-insensitive status message:

status = observer.wait_for_text(
    "#status", "complete", timeout=5, case_sensitive=False
)

wait_for_text_change(selector: str, previous: str | None = None, timeout: float | None = None) → str | None

Wait until element text differs from a supplied or current baseline.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
previous	str | None	Baseline text. None samples the current text when the element already exists.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
str | None	New element text when it differs from the baseline, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for a known status value to change:

new_text = observer.wait_for_text_change(
    "#status", previous="pending", timeout=5
)

wait_for_property(selector: str, property_name: str, expected: Any = <object object>, timeout: float | None = None) → Any | None

Wait for a DOM property to exist and optionally equal a value.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
property_name	str	JavaScript property name to inspect.
expected	Any	Exact expected value. When omitted, property existence is sufficient; explicitly passing None requires a null value.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider element reference when the property satisfies the request, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for a custom readiness property:

element = observer.wait_for_property(
    "#widget", "ready", True, timeout=5
)

wait_for_value(selector: str, expected: Any, timeout: float | None = None) → Any | None

Wait until an element’s DOM value property equals a value.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
expected	Any	Exact JSON-compatible value required from the DOM property.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider element reference when its value matches, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for an input value populated by JavaScript:

field = observer.wait_for_value("#email", "user@example.com")

wait_for_visible(selector: str, timeout: float | None = None) → Any | None

Wait for an element to exist and have a visible rendered box.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider element reference when visible, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for a dialog to become visible:

dialog = observer.wait_for_visible("[role='dialog']", timeout=5)

wait_for_hidden(selector: str, timeout: float | None = None) → bool

Wait for an element to be absent or not visibly rendered.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
bool	True when the element is absent or hidden; False on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for a progress overlay to stop blocking the page:

hidden = observer.wait_for_hidden(".progress-overlay", timeout=5)

wait_for_enabled(selector: str, timeout: float | None = None) → Any | None

Wait for an element to exist and not have a true disabled property.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider element reference when enabled, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait until submission becomes available:

button = observer.wait_for_enabled("button[type='submit']")

wait_for_disabled(selector: str, timeout: float | None = None) → Any | None

Wait for an element to exist and have a true disabled property.

Parameters

Name	Type	Description
selector	str	CSS selector resolved below the configured observer root.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider element reference when disabled, or None on timeout.

Raises

Exception	Description
ValueError	The selector is invalid or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait until a control becomes unavailable:

button = observer.wait_for_disabled("button[type='submit']")

wait_for_child_count(selector: str, *, count: int | None = None, minimum: int | None = None, maximum: int | None = None, timeout: float | None = None) → Any | None

Wait for a direct child count satisfying exact or bounded criteria.

Parameters

Name	Type	Description
selector	str	CSS selector for the parent below the observation root.
count	int | None	Required exact number of direct children, or None.
minimum	int | None	Inclusive minimum child count, or None.
maximum	int | None	Inclusive maximum child count, or None.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider parent-element reference when every supplied constraint is satisfied, or None on timeout.

Raises

Exception	Description
ValueError	No count constraint is supplied, the selector is invalid, or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for at least three direct list items:

list_element = observer.wait_for_child_count(
    "#results", minimum=3, timeout=5
)

wait_for_descendant_count(selector: str, descendant_selector: str, *, count: int | None = None, minimum: int | None = None, maximum: int | None = None, timeout: float | None = None) → Any | None

Wait for matching descendant count satisfying supplied constraints.

Parameters

Name	Type	Description
selector	str	CSS selector for the ancestor below the observation root.
descendant_selector	str	CSS selector counted within that ancestor.
count	int | None	Required exact number of descendants, or None.
minimum	int | None	Inclusive minimum descendant count, or None.
maximum	int | None	Inclusive maximum descendant count, or None.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Any | None	Provider ancestor-element reference when every supplied constraint is satisfied, or None on timeout.

Raises

Exception	Description
ValueError	No count constraint is supplied, either selector is invalid, or timeout is negative.
MutationObserverError	The observer is inactive or browser-side state becomes unavailable.

Examples

Wait for exactly two selected descendants:

panel = observer.wait_for_descendant_count(
    "#filters", "input:checked", count=2, timeout=5
)

wait_for_mutation(predicate: Callable[[MutationEvent], bool], timeout: float | None = None) → Tuple[MutationEvent, ...]

Wait for and return events accepted by a custom Python predicate.

Every batch is consumed from the browser buffer. Events rejected by the predicate are not restored.

Parameters

Name	Type	Description
predicate	Callable[[MutationEvent], bool]	Callable applied to each consumed event.
timeout	float | None	Maximum seconds to wait. None uses the default timeout.

Returns

Type	Description
Tuple[MutationEvent, …]	Matching events from the first successful batch, or an empty tuple on timeout or when observation becomes inactive.

Raises

Exception	Description
TypeError	predicate is not callable.
ValueError	The resolved timeout is negative.
MutationObserverError	Reading browser-side state fails.

Examples

Wait for attribute mutations only:

events = observer.wait_for_mutation(
    lambda event: event.type == "attributes",
    timeout=5,
)

has_records() → bool

Return whether at least one mutation is buffered without consuming it.

Returns

Type	Description
bool	True when the browser-side buffer contains at least one record; otherwise False.

Raises

Exception	Description
MutationObserverError	Browser-side state inspection fails.

Examples

Poll without draining the buffer:

pending = observer.has_records()

clear() → bool

Discard buffered events and reset the dropped-record counter.

Returns

Type	Description
bool	True when active browser-side state was cleared. False means the observer was inactive or its page state no longer existed.

Raises

Exception	Description
MutationObserverError	The browser rejects the clear command.

Examples

Discard observations collected during setup:

cleared = observer.clear()

is_active() → bool

Return whether browser-side state still exists on the current page.

Returns

Type	Description
bool	True when both Python and the current page retain active observer state; otherwise False. A false browser response marks this instance inactive.

Raises

Exception	Description
MutationObserverError	The browser rejects the state check.

Examples

Detect state lost after navigation:

active = observer.is_active()

stats() → Dict[str, Any]

Return active state plus buffered and dropped record counts.

Returns

Type	Description
Dict[str, Any]	A mapping containing active, buffered, and dropped. Inactive observers report zero buffered records while preserving the most recently observed dropped count.

Raises

Exception	Description
MutationObserverError	The browser rejects the statistics query.

Examples

Inspect buffer pressure without consuming events:

statistics = observer.stats()
buffered = statistics["buffered"]

property dropped_records: int

Return the most recently reported count of discarded old records.

Returns

Number reported by the most recent buffer response or statistics query. This cached value may lag behind browser-side state.

Examples

Read the cached overflow indicator:

dropped = observer.dropped_records

exception ddp_utils.browser.observers.mutation.MutationObserverError

Bases: RuntimeError

Base error for the Python-controlled browser MutationObserver.

Examples

Use this public operation:

instance = MutationObserverError()

exception ddp_utils.browser.observers.mutation.MutationObserverTargetError

Bases: MutationObserverError

Indicate that the requested CSS target does not exist on the page.

Examples

Use this public operation:

instance = MutationObserverTargetError()