Reference

Bases: ChromeDaemon

Source code in ichrome\daemon.py

class AsyncChromeDaemon(ChromeDaemon):
    _demo = r'''

    demo::

        import asyncio
        import json

        from ichrome import AsyncChromeDaemon


        async def main():
            async with AsyncChromeDaemon(clear_after_shutdown=True,
                                        headless=False,
                                        disable_image=False,
                                        user_data_dir='./ichrome_user_data') as cd:
                async with cd.connect_tab(0, auto_close=True) as tab:
                    loaded = await tab.goto('https://httpbin.org/forms/post',
                                            timeout=10)
                    html = await tab.html
                    title = await tab.title
                    print(
                        f'page loaded ok: {loaded}, HTML length is {len(html)}, title is "{title}"'
                    )
                    # try setting the input tag value with JS
                    await tab.js(
                        r"""document.querySelector('[value="bacon"]').checked = true""")
                    # or you can click the checkbox
                    await tab.click('[value="cheese"]')
                    # you can set the value of input
                    await tab.js(
                        r"""document.querySelector('[name="custname"]').value = "1234" """
                    )
                    # now click the submit button
                    await tab.click('form button')
                    await tab.wait_loading(5)
                    # extract the JSON with regex
                    result = await tab.findone(r'<pre.*?>([\s\S]*?)</pre>')
                    print(json.loads(result))


        if __name__ == "__main__":
            asyncio.run(main())

'''
    __doc__ = ChromeDaemon.__doc__ + _demo

    def __init__(
        self,
        chrome_path=None,
        host="127.0.0.1",
        port=9222,
        headless=False,
        user_agent=None,
        proxy=None,
        user_data_dir=None,
        disable_image=False,
        start_url="about:blank",
        extra_config=None,
        max_deaths=1,
        daemon=True,
        block=False,
        timeout=3,
        debug=False,
        proc_check_interval=5,
        on_startup=None,
        on_shutdown=None,
        before_startup=None,
        after_shutdown=None,
        clear_after_shutdown=False,
        popen_kwargs: dict = None,
    ):
        super().__init__(
            chrome_path=chrome_path,
            host=host,
            port=port,
            headless=headless,
            user_agent=user_agent,
            proxy=proxy,
            user_data_dir=user_data_dir,
            disable_image=disable_image,
            start_url=start_url,
            extra_config=extra_config,
            max_deaths=max_deaths,
            daemon=daemon,
            block=block,
            timeout=timeout,
            debug=debug,
            proc_check_interval=proc_check_interval,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
            before_startup=before_startup,
            after_shutdown=after_shutdown,
            clear_after_shutdown=clear_after_shutdown,
            popen_kwargs=popen_kwargs,
        )

    def init(self):
        # Please init AsyncChromeDaemon in a running loop with `async with`
        self._req = None
        self._chrome = AsyncChrome(self.host, self.port, timeout=self._timeout)
        self._init_coro = self._init_chrome_daemon()

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

    async def _init_chrome_daemon(self):
        await async_run(self._init_extra_config)
        await async_run(self._init_port)
        await async_run(self._wrap_user_data_dir)
        if not self.chrome_path:
            self.chrome_path = await async_run(self._get_default_path)
        _chrome_path = Path(self.chrome_path)
        if _chrome_path.is_file():
            CHROME_PROCESS_NAMES.add(_chrome_path.name)
        await async_run(self._ensure_port_free)
        self._req = Requests()
        if self.before_startup:
            await ensure_awaitable(self.before_startup(self))
        await self.launch_chrome()
        if self._use_daemon:
            self._daemon_thread = await self.run_forever(block=self._block)
        if self.on_startup:
            await ensure_awaitable(self.on_startup(self))
        return self

    async def restart(self):
        "restart the chrome process"
        logger.debug(f"restarting {self}")
        await async_run(self.kill)
        return await self.launch_chrome()

    async def launch_chrome(self):
        "launch the chrome with remote-debugging mode"
        await async_run(self._start_chrome_process)
        error = None
        for _ in range(int(self.MAX_WAIT_CHECKING_SECONDS * 2)):
            if not await async_run(self._proc_ok):
                error = "launch_chrome failed for proc not ok"
                break
            if await self._check_chrome_connection():
                self.ready = True
                break
            await asyncio.sleep(0.5)
        else:
            error = "launch_chrome failed for connection not ok"
        if error:
            logger.error(error)
            raise ChromeRuntimeError(error)

    async def _check_chrome_connection(self):
        r = await self.req.head(self.server, timeout=self._timeout)
        return r and r.ok

    async def check_connection(self):
        "check chrome connection ok"
        for _ in range(int(self.MAX_WAIT_CHECKING_SECONDS * 2)):
            if await self._check_chrome_connection():
                self.ready = True
                return True
            await asyncio.sleep(0.5)
        return False

    @property
    def connection_ok(self):
        return self.check_connection()

    @property
    def ok(self):
        return self.check_chrome_ready()

    @classmethod
    async def get_free_port(
        cls, host="127.0.0.1", start=9222, max_tries=100, timeout=1
    ):
        "find a free port which can be used"
        return await async_run(
            super().get_free_port,
            host=host,
            start=start,
            max_tries=max_tries,
            timeout=timeout,
        )

    async def check_ws_ready(self):
        async with self._chrome as chrome:
            return await chrome.check_ws_ready()

    async def check_chrome_ready(self):
        "check if the chrome api is available"
        if self.proc_ok and await self.check_connection():
            logger.debug(f"launch_chrome success: {self}, args: {self.proc.args}")
            return True
        else:
            logger.debug(f"launch_chrome failed: {self}, args: {self.cmd}")
            return False

    @property
    def loop(self):
        return asyncio.get_running_loop()

    async def run_forever(self, block=True, interval=None):
        "start the daemon and ensure proc is alive"
        interval = interval or self.proc_check_interval
        if self._shutdown:
            raise ChromeRuntimeError(
                f"{self} run_forever failed after shutdown({ttime(self._shutdown)})."
            )
        logger.debug(
            f"{self} run_forever(block={block}, interval={interval}, max_deaths={self.max_deaths})."
        )
        task = self._daemon_thread or asyncio.ensure_future(
            self._daemon(interval=interval)
        )
        if block:
            await task
        return task

    async def _daemon(self, interval=None):
        """if chrome proc is killed self.max_deaths times too fast (not raise TimeoutExpired),
        will skip auto_restart.
        check alive every `interval` seconds."""
        interval = interval or self.proc_check_interval
        return_code = None
        deaths = 0
        while self._use_daemon:
            if self._shutdown:
                logger.debug(
                    f"{self} daemon break after shutdown({ttime(self._shutdown)})."
                )
                break
            elif deaths >= self.max_deaths:
                logger.debug(
                    f"{self} daemon break for deaths is more than {self.max_deaths} times."
                )
                break
            elif not self.proc_ok:
                logger.debug(f"{self} daemon is restarting proc.")
                await self.restart()
                deaths += 1
                continue
            try:
                return_code = await async_run(self.proc.wait, interval)
                if self._shutdown_reason:
                    break
                deaths += 1
            except subprocess.TimeoutExpired:
                deaths = 0
        logger.debug(
            f"{self} daemon exited for {self._shutdown_reason}. return_code: {return_code}"
        )
        return return_code

    async def __aenter__(self):
        return await self._init_coro

    async def __aexit__(self, *args, **kwargs):
        await self.shutdown("__aexit__")

    @property
    def x(self):
        # `await self.x` to block until chrome daemon loop finished.
        if isawaitable(self._daemon_thread):
            return self._daemon_thread
        else:
            return asyncio.sleep(0)

    async def shutdown(self, reason=None):
        "shutdown the chrome, but do not use it, use async with instead."
        if self._shutdown:
            # logger.debug(f"{self} shutdown at {ttime(self._shutdown)} yet.")
            return
        self._shutdown_reason = reason
        self.update_shutdown_time()
        reason = f" for {reason}" if reason else ""
        logger.debug(
            f"{self} shutting down{reason}, start-up: {ttime(self.start_time)}, duration: {timepass(time.time() - self.start_time, accuracy=3, format=1)}."
        )
        if self.on_shutdown:
            await ensure_awaitable(self.on_shutdown(self))
        await async_run(self.kill, True)
        if self.after_shutdown:
            await ensure_awaitable(self.after_shutdown(self))
        if self.clear_after_shutdown:
            await self.clear_user_data_dir()

    async def _clear_user_dir(self):
        # Deprecated
        return await self.clear_user_data_dir()

    async def clear_user_data_dir(self):
        await self.shutdown("_clear_user_dir")
        return await async_run(self._clear_user_data_dir)

    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 chromed.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.

            View more about flatten: https://chromedevtools.github.io/devtools-protocol/tot/Target/#method-attachToTarget"""
        return _SingleTabConnectionManagerDaemon(
            host=self.host,
            port=self.port,
            index=index,
            auto_close=auto_close,
            flatten=flatten,
        )

    async def close_browser(self):
        "close browser peacefully"
        try:
            async with self.connect_tab(0) as tab:
                await tab.close_browser()
                return True
        except ChromeException:
            return False

    async def get_local_state(self):
        return await async_run(super().get_local_state)

    def create_context(
        self,
        disposeOnDetach: bool = True,
        proxyServer: str = None,
        proxyBypassList: str = None,
        originsWithUniversalNetworkAccess: List[str] = None,
    ) -> BrowserContext:
        "create a new browser context, which can be set new proxy, same like the incognito mode"
        return BrowserContext(
            chrome=AsyncChrome(host=self.host, port=self.port, timeout=self._timeout),
            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,
        disposeOnDetach: bool = True,
        proxyServer: str = None,
        proxyBypassList: str = None,
        originsWithUniversalNetworkAccess: List[str] = None,
        flatten: bool = None,
    ):
        "create a new tab with incognito mode, this is really a good choice"
        chrome = AsyncChrome(host=self.host, port=self.port, timeout=self._timeout)
        return chrome.incognito_tab(
            url=url,
            width=width,
            height=height,
            enableBeginFrameControl=enableBeginFrameControl,
            newWindow=newWindow,
            background=background,
            flatten=flatten,
            disposeOnDetach=disposeOnDetach,
            proxyServer=proxyServer,
            proxyBypassList=proxyBypassList,
            originsWithUniversalNetworkAccess=originsWithUniversalNetworkAccess,
        )

    def __del__(self):
        pass

check_chrome_ready() async

check if the chrome api is available

Source code in ichrome\daemon.py

async def check_chrome_ready(self):
    "check if the chrome api is available"
    if self.proc_ok and await self.check_connection():
        logger.debug(f"launch_chrome success: {self}, args: {self.proc.args}")
        return True
    else:
        logger.debug(f"launch_chrome failed: {self}, args: {self.cmd}")
        return False

check_connection() async

check chrome connection ok

Source code in ichrome\daemon.py

async def check_connection(self):
    "check chrome connection ok"
    for _ in range(int(self.MAX_WAIT_CHECKING_SECONDS * 2)):
        if await self._check_chrome_connection():
            self.ready = True
            return True
        await asyncio.sleep(0.5)
    return False

close_browser() async

close browser peacefully

Source code in ichrome\daemon.py

async def close_browser(self):
    "close browser peacefully"
    try:
        async with self.connect_tab(0) as tab:
            await tab.close_browser()
            return True
    except ChromeException:
        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 chromed.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.

View more about flatten: https://chromedevtools.github.io/devtools-protocol/tot/Target/#method-attachToTarget

Source code in ichrome\daemon.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 chromed.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.

        View more about flatten: https://chromedevtools.github.io/devtools-protocol/tot/Target/#method-attachToTarget"""
    return _SingleTabConnectionManagerDaemon(
        host=self.host,
        port=self.port,
        index=index,
        auto_close=auto_close,
        flatten=flatten,
    )

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

create a new browser context, which can be set new proxy, same like the incognito mode

Source code in ichrome\daemon.py

def create_context(
    self,
    disposeOnDetach: bool = True,
    proxyServer: str = None,
    proxyBypassList: str = None,
    originsWithUniversalNetworkAccess: List[str] = None,
) -> BrowserContext:
    "create a new browser context, which can be set new proxy, same like the incognito mode"
    return BrowserContext(
        chrome=AsyncChrome(host=self.host, port=self.port, timeout=self._timeout),
        disposeOnDetach=disposeOnDetach,
        proxyServer=proxyServer,
        proxyBypassList=proxyBypassList,
        originsWithUniversalNetworkAccess=originsWithUniversalNetworkAccess,
    )

get_free_port(host='127.0.0.1', start=9222, max_tries=100, timeout=1) async classmethod

find a free port which can be used

Source code in ichrome\daemon.py

@classmethod
async def get_free_port(
    cls, host="127.0.0.1", start=9222, max_tries=100, timeout=1
):
    "find a free port which can be used"
    return await async_run(
        super().get_free_port,
        host=host,
        start=start,
        max_tries=max_tries,
        timeout=timeout,
    )

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

create a new tab with incognito mode, this is really a good choice

Source code in ichrome\daemon.py

def incognito_tab(
    self,
    url: str = "about:blank",
    width: int = None,
    height: int = None,
    enableBeginFrameControl: bool = None,
    newWindow: bool = None,
    background: bool = None,
    disposeOnDetach: bool = True,
    proxyServer: str = None,
    proxyBypassList: str = None,
    originsWithUniversalNetworkAccess: List[str] = None,
    flatten: bool = None,
):
    "create a new tab with incognito mode, this is really a good choice"
    chrome = AsyncChrome(host=self.host, port=self.port, timeout=self._timeout)
    return chrome.incognito_tab(
        url=url,
        width=width,
        height=height,
        enableBeginFrameControl=enableBeginFrameControl,
        newWindow=newWindow,
        background=background,
        flatten=flatten,
        disposeOnDetach=disposeOnDetach,
        proxyServer=proxyServer,
        proxyBypassList=proxyBypassList,
        originsWithUniversalNetworkAccess=originsWithUniversalNetworkAccess,
    )

launch_chrome() async

launch the chrome with remote-debugging mode

Source code in ichrome\daemon.py

async def launch_chrome(self):
    "launch the chrome with remote-debugging mode"
    await async_run(self._start_chrome_process)
    error = None
    for _ in range(int(self.MAX_WAIT_CHECKING_SECONDS * 2)):
        if not await async_run(self._proc_ok):
            error = "launch_chrome failed for proc not ok"
            break
        if await self._check_chrome_connection():
            self.ready = True
            break
        await asyncio.sleep(0.5)
    else:
        error = "launch_chrome failed for connection not ok"
    if error:
        logger.error(error)
        raise ChromeRuntimeError(error)

restart() async

restart the chrome process

Source code in ichrome\daemon.py

async def restart(self):
    "restart the chrome process"
    logger.debug(f"restarting {self}")
    await async_run(self.kill)
    return await self.launch_chrome()

run_forever(block=True, interval=None) async

start the daemon and ensure proc is alive

Source code in ichrome\daemon.py

async def run_forever(self, block=True, interval=None):
    "start the daemon and ensure proc is alive"
    interval = interval or self.proc_check_interval
    if self._shutdown:
        raise ChromeRuntimeError(
            f"{self} run_forever failed after shutdown({ttime(self._shutdown)})."
        )
    logger.debug(
        f"{self} run_forever(block={block}, interval={interval}, max_deaths={self.max_deaths})."
    )
    task = self._daemon_thread or asyncio.ensure_future(
        self._daemon(interval=interval)
    )
    if block:
        await task
    return task

shutdown(reason=None) async

shutdown the chrome, but do not use it, use async with instead.

Source code in ichrome\daemon.py

async def shutdown(self, reason=None):
    "shutdown the chrome, but do not use it, use async with instead."
    if self._shutdown:
        # logger.debug(f"{self} shutdown at {ttime(self._shutdown)} yet.")
        return
    self._shutdown_reason = reason
    self.update_shutdown_time()
    reason = f" for {reason}" if reason else ""
    logger.debug(
        f"{self} shutting down{reason}, start-up: {ttime(self.start_time)}, duration: {timepass(time.time() - self.start_time, accuracy=3, format=1)}."
    )
    if self.on_shutdown:
        await ensure_awaitable(self.on_shutdown(self))
    await async_run(self.kill, True)
    if self.after_shutdown:
        await ensure_awaitable(self.after_shutdown(self))
    if self.clear_after_shutdown:
        await self.clear_user_data_dir()