Reference

Bases: GetValueMixin

Source code in ichrome\async_utils.py

class AsyncChrome(GetValueMixin):
    _DEFAULT_CONNECT_TIMEOUT = 3
    _DEFAULT_RETRY = 1

    def __init__(
        self,
        host: str = "127.0.0.1",
        port: int = 9222,
        timeout: int = None,
        retry: int = None,
    ):
        self.host = host
        # port can be null for chrome address without port.
        self.port = port
        self.timeout = timeout or self._DEFAULT_CONNECT_TIMEOUT
        self.retry = self._DEFAULT_RETRY if retry is None else retry
        self.status = "init"
        self._req = None
        self._browser: AsyncTab = None

    @property
    def browser(self) -> AsyncTab:
        if self._browser:
            return self._browser
        raise ChromeRuntimeError("`async with` context needed.")

    async def init_browser_tab(self):
        if self._browser:
            raise ChromeRuntimeError("`async with` context is already in use.")
        version = await self.version
        # print(version)
        self._browser = AsyncTab(
            tab_id="browser",
            type="browser",
            webSocketDebuggerUrl=version["webSocketDebuggerUrl"],
            chrome=self,
            flatten=False,
        )
        await self._browser.ws_connection.__aenter__()
        self.status = "connected"
        return self._browser

    def __getitem__(
        self, index: Union[int, str] = 0
    ) -> Awaitable[Union[AsyncTab, None]]:
        return self.get_tab(index=index)

    async def __aenter__(self):
        if await self.connect():
            await self.init_browser_tab()
        return self

    async def __aexit__(self, *args):
        await self.close()

    async def close(self):
        self.status = "disconnected"
        if self._req:
            await self._req.close()
        if self.status == "connected":
            await self._browser.ws_connection.__aexit__(None, None, None)
            self._browser = None

    async def close_browser(self):
        tab0 = await self.get_tab(0)
        if tab0:
            async with tab0():
                try:
                    await tab0.close_browser()
                except RuntimeError:
                    pass

    @property
    def server(self) -> str:
        """return like 'http://127.0.0.1:9222'"""
        if re.match(r"^https?://", self.host):
            prefix = self.host
        else:
            # filled the scheme
            prefix = f"http://{self.host}"
        if self.port:
            return f"{prefix}:{self.port}"
        else:
            return prefix

    async def get_version(self) -> dict:
        """`await self.get_version()`
        /json/version"""
        resp = await self.get_server("/json/version")
        if resp:
            return resp.json()
        else:
            return resp

    @property
    def version(self) -> Awaitable[dict]:
        """`await self.version`
        {'Browser': 'Chrome/77.0.3865.90', 'Protocol-Version': '1.3', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36', 'V8-Version': '7.7.299.11', 'WebKit-Version': '537.36 (@58c425ba843df2918d9d4b409331972646c393dd)', 'webSocketDebuggerUrl': 'ws://127.0.0.1:9222/devtools/browser/b5fbd149-959b-4603-b209-cfd26d66bdc1'}"""
        return self.get_version()

    @property
    def meta(self):
        # same for Sync Chrome
        return self.get_version()

    async def connect(self) -> bool:
        """await self.connect()"""
        self._req = Requests()
        if await self.check_http_ready():
            return True
        else:
            return False

    @property
    def req(self) -> Requests:
        if self._req is None:
            raise ChromeRuntimeError("please use Chrome in `async with` context")
        return self._req

    async def check(self) -> bool:
        """Test http connection to cdp. `await self.check()`"""
        return bool(await self.check_http_ready()) and (await self.check_ws_ready())

    async def check_http_ready(self):
        resp = await self.req.head(self.server, timeout=self.timeout, retry=self.retry)
        if not resp:
            self.status = resp.text
        return bool(resp)

    async def check_ws_ready(self):
        try:
            data = await self.browser.browser_version()
            return bool(isinstance(data, dict) and data.get("result"))
        except Exception:
            return False

    @property
    def ok(self):
        """await self.ok"""
        return self.check()

    async def get_server(self, api: str = "", method="GET") -> NewResponse:
        # maybe return failure request
        url = urljoin(self.server, api)
        resp = await self.req.request(
            method=method, url=url, timeout=self.timeout, retry=self.retry
        )
        if not resp:
            self.status = resp.text
        return resp

    async def get_tabs(self, filt_page_type: bool = True) -> List[AsyncTab]:
        """`await self.get_tabs()`.
        cdp url: /json"""
        try:
            r = await self.get_server("/json")
            if r:
                return [
                    AsyncTab(chrome=self, json=rjson, **rjson)
                    for rjson in r.json()
                    if (rjson["type"] == "page" or filt_page_type is not True)
                ]
        except Exception as error:
            logger.error(f"fail to get_tabs from {self.server}, {error!r}")
            raise error
        return []

    async def get_tab(self, index: Union[int, str] = 0) -> Union[AsyncTab, None]:
        """`await self.get_tab(1)` <=> await `(await self.get_tabs())[1]`
        If not exist, return None
        cdp url: /json"""
        tabs = await self.get_tabs()
        try:
            if isinstance(index, int):
                return tabs[index]
            else:
                for tab in tabs:
                    if tab.tab_id == index:
                        return tab
        except IndexError:
            pass
        return None

    @property
    def tabs(self) -> Awaitable[List[AsyncTab]]:
        """`await self.tabs`. tabs[0] is the current activated tab"""
        # [{'description': '', 'devtoolsFrontendUrl': '/devtools/inspector.html?ws=127.0.0.1:9222/devtools/page/30C16F9165C525A4002E827EDABD48A4', 'id': '30C16F9165C525A4002E827EDABD48A4', 'title': 'about:blank', 'type': 'page', 'url': 'about:blank', 'webSocketDebuggerUrl': 'ws://127.0.0.1:9222/devtools/page/30C16F9165C525A4002E827EDABD48A4'}]
        return self.get_tabs()

    async def kill(
        self, timeout: Union[int, float] = None, max_deaths: int = 1
    ) -> None:
        if self.req:
            await self.req.close()
        await async_run(
            clear_chrome_process, self.port, timeout, max_deaths, host=self.host
        )

    async def new_tab(self, url: str = "") -> Union[AsyncTab, None]:
        api = f"/json/new?{quote_plus(url)}"
        r = await self.get_server(api, method="PUT")
        if r:
            rjson = r.json()
            tab = AsyncTab(chrome=self, **rjson)
            tab._created_time = int(time.time())
            logger.debug(f"[new_tab] {tab} {rjson}")
            return tab
        else:
            return None

    async def do_tab(
        self, tab_id: Union[AsyncTab, str], action: str
    ) -> Union[str, bool]:
        ok = False
        if isinstance(tab_id, AsyncTab):
            tab_id = tab_id.tab_id
        r = await self.get_server(f"/json/{action}/{tab_id}")
        if r:
            if action == "close":
                ok = r.text == "Target is closing"
            elif action == "activate":
                ok = r.text == "Target activated"
            else:
                ok == r.text
        logger.debug(f"[{action}_tab] <Tab: {tab_id}>: {ok}")
        return ok

    async def activate_tab(self, tab_id: Union[AsyncTab, str]) -> Union[str, bool]:
        return await self.do_tab(tab_id, action="activate")

    async def close_tab(self, tab_id: Union[AsyncTab, str]) -> Union[str, bool]:
        return await self.do_tab(tab_id, action="close")

    async def close_tabs(
        self, tab_ids: Union[None, List[AsyncTab], List[str]] = None, *args
    ) -> List[Union[str, bool]]:
        if tab_ids is None:
            tab_ids = await self.tabs
        return [await self.close_tab(tab_id) for tab_id in tab_ids]

    def connect_tab(
        self,
        index: Union[None, int, str] = 0,
        auto_close: bool = False,
        flatten: bool = None,
    ):
        """More easier way to init a connected Tab with `async with`.

        Got a connected Tab object by using `async with chrome.connect_tab(0) as tab::`

            index = 0 means the current tab.
            index = None means create a new tab.
            index = 'http://python.org' means create a new tab with url.
            index = 'F130D0295DB5879791AA490322133AFC' means the tab with this id.

            If auto_close is True: close this tab while exiting context."""
        return _SingleTabConnectionManager(
            chrome=self, index=index, auto_close=auto_close, flatten=flatten
        )

    def connect_tabs(self, *tabs) -> "_TabConnectionManager":
        """async with chrome.connect_tabs([tab1, tab2]):.
        or
        async with chrome.connect_tabs(tab1, tab2)"""
        if not tabs:
            raise ChromeValueError(
                f"tabs should not be null, but `{tabs!r}` was given."
            )
        tab0 = tabs[0]
        if isinstance(tab0, (list, tuple)):
            tabs_todo = tab0
        else:
            tabs_todo = tabs
        return _TabConnectionManager(tabs_todo)

    def get_memory(self, attr="uss", unit="MB"):
        """Only support local Daemon. `uss` is slower than `rss` but useful."""
        return get_memory_by_port(port=self.port, attr=attr, unit=unit, host=self.host)

    def __repr__(self):
        return f"<Chrome({self.status}): {self.port}>"

    def __str__(self):
        return f"<Chrome({self.status}): {self.server}>"

    def __del__(self):
        _exhaust_simple_coro(self.close())

    def create_context(
        self,
        disposeOnDetach: bool = True,
        proxyServer: str = None,
        proxyBypassList: str = None,
        originsWithUniversalNetworkAccess: List[str] = None,
    ) -> "BrowserContext":
        "create a new Incognito BrowserContext"
        return BrowserContext(
            chrome=self,
            disposeOnDetach=disposeOnDetach,
            proxyServer=proxyServer,
            proxyBypassList=proxyBypassList,
            originsWithUniversalNetworkAccess=originsWithUniversalNetworkAccess,
        )

    def incognito_tab(
        self,
        url: str = "about:blank",
        width: int = None,
        height: int = None,
        enableBeginFrameControl: bool = None,
        newWindow: bool = None,
        background: bool = None,
        flatten: bool = None,
        disposeOnDetach: bool = True,
        proxyServer: str = None,
        proxyBypassList: str = None,
        originsWithUniversalNetworkAccess: List[str] = None,
    ) -> "IncognitoTabContext":
        "create a new Incognito tab"
        return IncognitoTabContext(
            chrome=self,
            url=url,
            width=width,
            height=height,
            enableBeginFrameControl=enableBeginFrameControl,
            newWindow=newWindow,
            background=background,
            flatten=flatten,
            disposeOnDetach=disposeOnDetach,
            proxyServer=proxyServer,
            proxyBypassList=proxyBypassList,
            originsWithUniversalNetworkAccess=originsWithUniversalNetworkAccess,
        )

check() async

Test http connection to cdp. await self.check()

Source code in ichrome\async_utils.py

async def check(self) -> bool:
    """Test http connection to cdp. `await self.check()`"""
    return bool(await self.check_http_ready()) and (await self.check_ws_ready())

connect() async

await self.connect()

Source code in ichrome\async_utils.py

async def connect(self) -> bool:
    """await self.connect()"""
    self._req = Requests()
    if await self.check_http_ready():
        return True
    else:
        return False

connect_tab(index=0, auto_close=False, flatten=None)

More easier way to init a connected Tab with async with.

Got a connected Tab object by using async with chrome.connect_tab(0) as tab::

index = 0 means the current tab.
index = None means create a new tab.
index = 'http://python.org' means create a new tab with url.
index = 'F130D0295DB5879791AA490322133AFC' means the tab with this id.

If auto_close is True: close this tab while exiting context.

Source code in ichrome\async_utils.py

def connect_tab(
    self,
    index: Union[None, int, str] = 0,
    auto_close: bool = False,
    flatten: bool = None,
):
    """More easier way to init a connected Tab with `async with`.

    Got a connected Tab object by using `async with chrome.connect_tab(0) as tab::`

        index = 0 means the current tab.
        index = None means create a new tab.
        index = 'http://python.org' means create a new tab with url.
        index = 'F130D0295DB5879791AA490322133AFC' means the tab with this id.

        If auto_close is True: close this tab while exiting context."""
    return _SingleTabConnectionManager(
        chrome=self, index=index, auto_close=auto_close, flatten=flatten
    )

connect_tabs(*tabs)

async with chrome.connect_tabs([tab1, tab2]):. or async with chrome.connect_tabs(tab1, tab2)

Source code in ichrome\async_utils.py

def connect_tabs(self, *tabs) -> "_TabConnectionManager":
    """async with chrome.connect_tabs([tab1, tab2]):.
    or
    async with chrome.connect_tabs(tab1, tab2)"""
    if not tabs:
        raise ChromeValueError(
            f"tabs should not be null, but `{tabs!r}` was given."
        )
    tab0 = tabs[0]
    if isinstance(tab0, (list, tuple)):
        tabs_todo = tab0
    else:
        tabs_todo = tabs
    return _TabConnectionManager(tabs_todo)

create_context(disposeOnDetach=True, proxyServer=None, proxyBypassList=None, originsWithUniversalNetworkAccess=None)

create a new Incognito BrowserContext

Source code in ichrome\async_utils.py

def create_context(
    self,
    disposeOnDetach: bool = True,
    proxyServer: str = None,
    proxyBypassList: str = None,
    originsWithUniversalNetworkAccess: List[str] = None,
) -> "BrowserContext":
    "create a new Incognito BrowserContext"
    return BrowserContext(
        chrome=self,
        disposeOnDetach=disposeOnDetach,
        proxyServer=proxyServer,
        proxyBypassList=proxyBypassList,
        originsWithUniversalNetworkAccess=originsWithUniversalNetworkAccess,
    )

get_memory(attr='uss', unit='MB')

Only support local Daemon. uss is slower than rss but useful.

Source code in ichrome\async_utils.py

def get_memory(self, attr="uss", unit="MB"):
    """Only support local Daemon. `uss` is slower than `rss` but useful."""
    return get_memory_by_port(port=self.port, attr=attr, unit=unit, host=self.host)

get_tab(index=0) async

await self.get_tab(1) <=> await (await self.get_tabs())[1] If not exist, return None cdp url: /json

Source code in ichrome\async_utils.py

async def get_tab(self, index: Union[int, str] = 0) -> Union[AsyncTab, None]:
    """`await self.get_tab(1)` <=> await `(await self.get_tabs())[1]`
    If not exist, return None
    cdp url: /json"""
    tabs = await self.get_tabs()
    try:
        if isinstance(index, int):
            return tabs[index]
        else:
            for tab in tabs:
                if tab.tab_id == index:
                    return tab
    except IndexError:
        pass
    return None

get_tabs(filt_page_type=True) async

await self.get_tabs(). cdp url: /json

Source code in ichrome\async_utils.py

async def get_tabs(self, filt_page_type: bool = True) -> List[AsyncTab]:
    """`await self.get_tabs()`.
    cdp url: /json"""
    try:
        r = await self.get_server("/json")
        if r:
            return [
                AsyncTab(chrome=self, json=rjson, **rjson)
                for rjson in r.json()
                if (rjson["type"] == "page" or filt_page_type is not True)
            ]
    except Exception as error:
        logger.error(f"fail to get_tabs from {self.server}, {error!r}")
        raise error
    return []

get_version() async

await self.get_version() /json/version

Source code in ichrome\async_utils.py

async def get_version(self) -> dict:
    """`await self.get_version()`
    /json/version"""
    resp = await self.get_server("/json/version")
    if resp:
        return resp.json()
    else:
        return resp

incognito_tab(url='about:blank', width=None, height=None, enableBeginFrameControl=None, newWindow=None, background=None, flatten=None, disposeOnDetach=True, proxyServer=None, proxyBypassList=None, originsWithUniversalNetworkAccess=None)

create a new Incognito tab

Source code in ichrome\async_utils.py

def incognito_tab(
    self,
    url: str = "about:blank",
    width: int = None,
    height: int = None,
    enableBeginFrameControl: bool = None,
    newWindow: bool = None,
    background: bool = None,
    flatten: bool = None,
    disposeOnDetach: bool = True,
    proxyServer: str = None,
    proxyBypassList: str = None,
    originsWithUniversalNetworkAccess: List[str] = None,
) -> "IncognitoTabContext":
    "create a new Incognito tab"
    return IncognitoTabContext(
        chrome=self,
        url=url,
        width=width,
        height=height,
        enableBeginFrameControl=enableBeginFrameControl,
        newWindow=newWindow,
        background=background,
        flatten=flatten,
        disposeOnDetach=disposeOnDetach,
        proxyServer=proxyServer,
        proxyBypassList=proxyBypassList,
        originsWithUniversalNetworkAccess=originsWithUniversalNetworkAccess,
    )

ok() property

await self.ok

Source code in ichrome\async_utils.py

@property
def ok(self):
    """await self.ok"""
    return self.check()

server() property

return like 'http://127.0.0.1:9222'

Source code in ichrome\async_utils.py

@property
def server(self) -> str:
    """return like 'http://127.0.0.1:9222'"""
    if re.match(r"^https?://", self.host):
        prefix = self.host
    else:
        # filled the scheme
        prefix = f"http://{self.host}"
    if self.port:
        return f"{prefix}:{self.port}"
    else:
        return prefix

tabs() property

await self.tabs. tabs[0] is the current activated tab

Source code in ichrome\async_utils.py

@property
def tabs(self) -> Awaitable[List[AsyncTab]]:
    """`await self.tabs`. tabs[0] is the current activated tab"""
    # [{'description': '', 'devtoolsFrontendUrl': '/devtools/inspector.html?ws=127.0.0.1:9222/devtools/page/30C16F9165C525A4002E827EDABD48A4', 'id': '30C16F9165C525A4002E827EDABD48A4', 'title': 'about:blank', 'type': 'page', 'url': 'about:blank', 'webSocketDebuggerUrl': 'ws://127.0.0.1:9222/devtools/page/30C16F9165C525A4002E827EDABD48A4'}]
    return self.get_tabs()

version() property

await self.version {'Browser': 'Chrome/77.0.3865.90', 'Protocol-Version': '1.3', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36', 'V8-Version': '7.7.299.11', 'WebKit-Version': '537.36 (@58c425ba843df2918d9d4b409331972646c393dd)', 'webSocketDebuggerUrl': 'ws://127.0.0.1:9222/devtools/browser/b5fbd149-959b-4603-b209-cfd26d66bdc1'}

Source code in ichrome\async_utils.py

@property
def version(self) -> Awaitable[dict]:
    """`await self.version`
    {'Browser': 'Chrome/77.0.3865.90', 'Protocol-Version': '1.3', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36', 'V8-Version': '7.7.299.11', 'WebKit-Version': '537.36 (@58c425ba843df2918d9d4b409331972646c393dd)', 'webSocketDebuggerUrl': 'ws://127.0.0.1:9222/devtools/browser/b5fbd149-959b-4603-b209-cfd26d66bdc1'}"""
    return self.get_version()