Skip to content

Session

SessionModule

Module for managing the BiDi session.

Source code in bidiwave/modules/session.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class SessionModule:
    """Module for managing the BiDi session."""

    def __init__(self, connection: Connection) -> None:
        self._connection = connection
        # Called with the session.new result so the client can pick up
        # the negotiated capabilities.
        self.on_new: Callable[[dict[str, Any]], None] | None = None

    async def new(self, capabilities: dict[str, Any] | None = None) -> Session:
        params = {
            "capabilities": {
                "alwaysMatch": capabilities if capabilities is not None else {"webSocketUrl": True},
            }
        }
        result = await self._connection.send_command(SESSION_NEW, params)
        if self.on_new is not None:
            self.on_new(result)
        return Session.model_validate(result)

    async def status(self) -> SessionStatus:
        result = await self._connection.send_command(SESSION_STATUS, {})
        return SessionStatus.model_validate(result)

    async def end(self) -> None:
        """Closes the current BiDi session."""
        await self._connection.send_command(SESSION_END, {})

    async def subscribe(
        self,
        events: list[str],
        contexts: list[str] | None = None,
        user_contexts: list[str] | None = None,
    ) -> dict[str, Any]:
        """Subscribes to browser events.

        Args:
            events: Event names to subscribe to.
            contexts: Scope to top-level browsing context IDs.
            user_contexts: Scope to user context IDs.

        Returns:
            Subscription result from the server (contains "subscription"
            with the subscription ID on supporting implementations).
        """
        params: dict[str, Any] = {"events": events}
        if contexts is not None:
            params["contexts"] = contexts
        if user_contexts is not None:
            params["userContexts"] = user_contexts
        return await self._connection.send_command(SESSION_SUBSCRIBE, params)

    async def unsubscribe(
        self,
        events: list[str] | None = None,
        contexts: list[str] | None = None,
        subscriptions: list[str] | None = None,
    ) -> None:
        """Unsubscribes from events.

        Either `events` (+ optional `contexts`) or `subscriptions` must
        be given — the two forms are mutually exclusive per the spec.
        """
        if subscriptions is not None:
            params: dict[str, Any] = {"subscriptions": subscriptions}
        else:
            if events is None:
                raise ValueError("events or subscriptions must be provided")
            params = {"events": events}
            if contexts is not None:
                params["contexts"] = contexts
        await self._connection.send_command(SESSION_UNSUBSCRIBE, params)

end async 

end() -> None

Closes the current BiDi session.

Source code in bidiwave/modules/session.py 
41
42
43
async def end(self) -> None:
    """Closes the current BiDi session."""
    await self._connection.send_command(SESSION_END, {})

subscribe async 

subscribe(events: list[str], contexts: list[str] | None = None, user_contexts: list[str] | None = None) -> dict[str, Any]

Subscribes to browser events.

Parameters:

Name Type Description Default
events list[str]

Event names to subscribe to.

 required
contexts list[str] | None

Scope to top-level browsing context IDs.

 None
user_contexts list[str] | None

Scope to user context IDs.

 None

Returns:

Type Description
dict[str, Any]

Subscription result from the server (contains "subscription"
dict[str, Any]

with the subscription ID on supporting implementations).
Source code in bidiwave/modules/session.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
async def subscribe(
    self,
    events: list[str],
    contexts: list[str] | None = None,
    user_contexts: list[str] | None = None,
) -> dict[str, Any]:
    """Subscribes to browser events.

    Args:
        events: Event names to subscribe to.
        contexts: Scope to top-level browsing context IDs.
        user_contexts: Scope to user context IDs.

    Returns:
        Subscription result from the server (contains "subscription"
        with the subscription ID on supporting implementations).
    """
    params: dict[str, Any] = {"events": events}
    if contexts is not None:
        params["contexts"] = contexts
    if user_contexts is not None:
        params["userContexts"] = user_contexts
    return await self._connection.send_command(SESSION_SUBSCRIBE, params)

unsubscribe async 

unsubscribe(events: list[str] | None = None, contexts: list[str] | None = None, subscriptions: list[str] | None = None) -> None

Unsubscribes from events.

Either events (+ optional contexts) or subscriptions must be given — the two forms are mutually exclusive per the spec.

Source code in bidiwave/modules/session.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def unsubscribe(
    self,
    events: list[str] | None = None,
    contexts: list[str] | None = None,
    subscriptions: list[str] | None = None,
) -> None:
    """Unsubscribes from events.

    Either `events` (+ optional `contexts`) or `subscriptions` must
    be given — the two forms are mutually exclusive per the spec.
    """
    if subscriptions is not None:
        params: dict[str, Any] = {"subscriptions": subscriptions}
    else:
        if events is None:
            raise ValueError("events or subscriptions must be provided")
        params = {"events": events}
        if contexts is not None:
            params["contexts"] = contexts
    await self._connection.send_command(SESSION_UNSUBSCRIBE, params)