Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859"""Subscriptions are explicit scopes, separate from observation and interaction."""
from __future__ import annotationsfrom typing import Anyfrom .transport import CDPSession, Subscription
class EventScope: def __init__( self, cdp: CDPSession, method: str, *, capacity: int = 256, max_bytes: int = 2 * 1024 * 1024, enable: bool = True, ) -> None: self.cdp = cdp self.method = method self.capacity = capacity self.max_bytes = max_bytes self.enable = enable self.subscription: Subscription | None = None
async def __aenter__(self) -> Subscription: if self.subscription is not None: raise RuntimeError("An event scope cannot be entered twice") self.subscription = self.cdp.connection.subscribe( self.method, self.cdp.session_id, capacity=self.capacity, max_bytes=self.max_bytes, ) try: if self.enable: domain = self.method.split(".")[0] if domain not in { "Network", "Runtime", "Page", "DOM", "Accessibility", "Log", "Performance", "Security", }: raise ValueError( f"Enable {domain} explicitly with CDP, then use enable=False" ) await self.cdp.send(f"{domain}.enable") return self.subscription except BaseException: self.subscription.close() raise
async def __aexit__(self, *exc: Any) -> None: if self.subscription is not None: self.subscription.close()