ddp_utils.browser.facade.events

import ddp_utils.browser.facade.events

Shared synchronous browser event subscription contract.

class ddp_utils.browser.facade.events.BrowserSubscription(event: str, callback: ~typing.Callable[[...], ~typing.Any], native: ~typing.Any = None, identifier: str = <factory>, active: bool = True, details: dict[str, ~typing.Any] = <factory>)

Bases: object

Represent one removable browser event subscription.

Parameters

Name	Type	Description
event	str	Stable or provider-native event name.
callback	Callable[[...], Any]	User callback receiving the normalized or native event.
native	Any	Provider-specific listener or route handle.
identifier	str	Process-local stable subscription identifier.
active	bool	Whether the subscription is still installed.
details	dict[str, Any]	Additional non-secret diagnostics.

Examples

Remove a listener through its owning service:

subscription = browser.cdp.on("Network.responseReceived", callback)
browser.cdp.off(subscription.require())

deactivate() → None

Mark the subscription inactive after provider cleanup.

Examples

Owning services deactivate only after listener removal:

subscription.deactivate()

class ddp_utils.browser.facade.events.BrowserWatcher(name: str, when: dict[str, ~typing.Any], callback: ~typing.Callable[[~ddp_utils.browser.facade.events.BrowserWatcherEvent], ~typing.Any], once: bool = False, cooldown: float = 0.0, trigger_existing: bool = True, identifier: str = <factory>, active: bool = True)

Bases: object

Describe one active browser-state watcher.

Parameters

Name	Type	Description
name	str	Unique application-facing watcher name.
when	dict[str, Any]	Serializable condition evaluated inside the browser page.
callback	Callable[[BrowserWatcherEvent], Any]	Callable receiving each matching BrowserWatcherEvent.
once	bool	Remove the watcher before its first callback is delivered.
cooldown	float	Minimum seconds between repeated false-to-true transitions.
trigger_existing	bool	Emit when the condition is already true at registration.
identifier	str	Process-local stable watcher identifier.
active	bool	Whether the watcher remains registered.

Examples

Inspect a watcher returned by browser.events.add_watcher:

watcher = browser.events.add_watcher(
    name="confirmation",
    when={"css": ".confirmation", "condition": "visible"},
    callback=handle_confirmation,
)
assert watcher.active

deactivate() → None

Mark the watcher inactive after removal.

Examples

Owning event services deactivate removed watchers:

watcher.deactivate()

runtime_payload() → dict[str, Any]

Return the secret-free browser-side watcher representation.

Returns

Type	Description
dict[str, Any]	Serializable mapping consumed by the shared JavaScript runtime.

Examples

Synchronize the watcher without exposing its Python callback:

payload = watcher.runtime_payload()

class ddp_utils.browser.facade.events.BrowserWatcherEvent(name: str, watcher_id: str, occurred_at: float, sequence: int, url: str, when: Mapping[str, Any], browser: Any, source: str = 'dom', payload: Any = None)

Bases: object

Represent one normalized browser condition transition.

Parameters

Name	Type	Description
name	str	Application-facing watcher name.
watcher_id	str	Stable identifier of the matching watcher.
occurred_at	float	Browser wall-clock timestamp in seconds.
sequence	int	Monotonic sequence within the current document runtime.
url	str	Page URL observed when the condition matched.
when	Mapping[str, Any]	Condition registered for the watcher.
browser	Any	Owning browser facade available to the callback.
source	str	Observation source, currently "dom" or "network".
payload	Any	Source-specific normalized payload. Network events carry a NetworkRecord; DOM events use None.

Examples

Use the owning facade from a synchronous callback:

def handle_confirmation(event):
    event.browser.find("button", text="Confirm").click()

Inspect a normalized network response without backend branching:

def handle_api(event):
    print(event.payload.status, event.payload.url)