82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
|
"""
|
||
|
|
This type stub file was generated by pyright.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
from typing import Awaitable, Callable, Dict, TYPE_CHECKING, Union
|
||
|
|
from pyee import EventEmitter
|
||
|
|
|
||
|
|
"""Connection/Session management module."""
|
||
|
|
if TYPE_CHECKING:
|
||
|
|
...
|
||
|
|
logger = ...
|
||
|
|
logger_connection = ...
|
||
|
|
logger_session = ...
|
||
|
|
class Connection(EventEmitter):
|
||
|
|
"""Connection management class."""
|
||
|
|
def __init__(self, url: str, loop: asyncio.AbstractEventLoop, delay: int = ...) -> None:
|
||
|
|
"""Make connection.
|
||
|
|
|
||
|
|
:arg str url: WebSocket url to connect devtool.
|
||
|
|
:arg int delay: delay to wait before processing received messages.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def url(self) -> str:
|
||
|
|
"""Get connected WebSocket url."""
|
||
|
|
...
|
||
|
|
|
||
|
|
def send(self, method: str, params: dict = ...) -> Awaitable:
|
||
|
|
"""Send message via the connection."""
|
||
|
|
...
|
||
|
|
|
||
|
|
def setClosedCallback(self, callback: Callable[[], None]) -> None:
|
||
|
|
"""Set closed callback."""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def dispose(self) -> None:
|
||
|
|
"""Close all connection."""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def createSession(self, targetInfo: Dict) -> CDPSession:
|
||
|
|
"""Create new session."""
|
||
|
|
...
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
class CDPSession(EventEmitter):
|
||
|
|
"""Chrome Devtools Protocol Session.
|
||
|
|
|
||
|
|
The :class:`CDPSession` instances are used to talk raw Chrome Devtools
|
||
|
|
Protocol:
|
||
|
|
|
||
|
|
* protocol methods can be called with :meth:`send` method.
|
||
|
|
* protocol events can be subscribed to with :meth:`on` method.
|
||
|
|
|
||
|
|
Documentation on DevTools Protocol can be found
|
||
|
|
`here <https://chromedevtools.github.io/devtools-protocol/>`__.
|
||
|
|
"""
|
||
|
|
def __init__(self, connection: Union[Connection, CDPSession], targetType: str, sessionId: str, loop: asyncio.AbstractEventLoop) -> None:
|
||
|
|
"""Make new session."""
|
||
|
|
...
|
||
|
|
|
||
|
|
def send(self, method: str, params: dict = ...) -> Awaitable:
|
||
|
|
"""Send message to the connected session.
|
||
|
|
|
||
|
|
:arg str method: Protocol method name.
|
||
|
|
:arg dict params: Optional method parameters.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def detach(self) -> None:
|
||
|
|
"""Detach session from target.
|
||
|
|
|
||
|
|
Once detached, session won't emit any events and can't be used to send
|
||
|
|
messages.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
|
||
|
|
|