How to use handle_transport_close method in Playwright Python

1import asyncio
2import importlib.metadata
3from typing import Any, Dict, cast
4
5from greenlet import greenlet
6
7from playwright._impl._api_types import Error
8from playwright._impl._browser import Browser as BrowserImpl
9from playwright._impl._connection import Connection, from_channel
10from playwright._impl._object_factory import create_remote_object
11from playwright._impl._playwright import Playwright as PlaywrightImpl
12from playwright._impl._transport import WebSocketTransport
13from playwright.sync_api._generated import Browser as SyncBrowser
14
15
16class ConnectToBrowserContextManager:
17    def __init__(
18        self,
19        ws_endpoint: str,
20        slow_mo: float = None,
21        headers: Dict[str, str] = None,
22    ) -> None:
23        self._browser: SyncBrowser
24        self._ws_endpoint = ws_endpoint
25        self._slow_mo = slow_mo
26        self._headers = headers
27
28    def _make_connection(self, dispatcher_fiber, object_factory, transport, loop) -> Connection:
29        if importlib.metadata.version('playwright') < '1.15.0':
30            return Connection(
31                dispatcher_fiber,
32                create_remote_object,
33                transport)
34        else:
35            return Connection(
36                dispatcher_fiber,
37                create_remote_object,
38                transport,
39                loop)
40
41    def _setup_browser(self, browser: BrowserImpl):
42        version = importlib.metadata.version('playwright')
43        if version < '1.17.0':
44            browser._is_remote = True
45            browser._is_connected_over_websocket = True
46        elif version < '1.19.0':
47            browser._should_close_connection_on_close = True
48        else:
49            browser._should_close_connection_on_close = True
50            browser._local_utils = self._playwright_impl._utils
51
52    def __enter__(self) -> SyncBrowser:
53        loop: asyncio.AbstractEventLoop
54        own_loop = None
55        try:
56            loop = asyncio.get_running_loop()
57        except RuntimeError:
58            loop = asyncio.new_event_loop()
59            own_loop = loop
60        if loop.is_running():
61            raise Error(
62                """It looks like you are using Playwright Sync API inside the asyncio loop.
63Please use the Async API instead."""
64            )
65
66        def greenlet_main() -> None:
67            loop.run_until_complete(self._connection.run_as_sync())
68
69            if own_loop:
70                loop.run_until_complete(loop.shutdown_asyncgens())
71                loop.close()
72
73        dispatcher_fiber = greenlet(greenlet_main)
74        transport = WebSocketTransport(
75            loop,
76            self._ws_endpoint,
77            self._headers,
78            self._slow_mo)
79        self._connection = self._make_connection(
80            dispatcher_fiber,
81            create_remote_object,
82            transport,
83            loop)
84
85        g_self = greenlet.getcurrent()
86
87        def callback_wrapper(playwright_impl: PlaywrightImpl) -> None:
88            self._playwright_impl = playwright_impl
89            g_self.switch()
90
91        self._connection.call_on_object_with_known_name(
92            "Playwright", callback_wrapper)
93
94        dispatcher_fiber.switch()
95
96        pre_launched_browser = self._playwright_impl._initializer.get(
97            "preLaunchedBrowser")
98        assert pre_launched_browser
99        browser = cast(BrowserImpl, from_channel(pre_launched_browser))
100        self._setup_browser(browser)
101
102        def handle_transport_close() -> None:
103            for context in browser.contexts:
104                for page in context.pages:
105                    page._on_close()
106                context._on_close()
107            browser._on_close()
108
109        transport.once("close", handle_transport_close)
110
111        self._browser = SyncBrowser(browser)
112        return self._browser
113
114    def start(self) -> SyncBrowser:
115        return self.__enter__()
116
117    def __exit__(self, *args: Any) -> None:
118        self._browser.close()
119
120
121def connect_to_browser(
122    ws_endpoint: str,
123    slow_mo: float = None,
124    headers: Dict[str, str] = None,
125) -> SyncBrowser:
126    return ConnectToBrowserContextManager(ws_endpoint, slow_mo, headers)
127

Source: connect_to_browser.py

import asyncio
import importlib.metadata
from typing import Any, Dict, cast

from greenlet import greenlet

from playwright._impl._api_types import Error
from playwright._impl._browser import Browser as BrowserImpl
from playwright._impl._connection import Connection, from_channel
from playwright._impl._object_factory import create_remote_object
from playwright._impl._playwright import Playwright as PlaywrightImpl
from playwright._impl._transport import WebSocketTransport
from playwright.sync_api._generated import Browser as SyncBrowser


class ConnectToBrowserContextManager:
    def __init__(
        self,
        ws_endpoint: str,
        slow_mo: float = None,
        headers: Dict[str, str] = None,
    ) -> None:
        self._browser: SyncBrowser
        self._ws_endpoint = ws_endpoint
        self._slow_mo = slow_mo
        self._headers = headers

    def _make_connection(self, dispatcher_fiber, object_factory, transport, loop) -> Connection:
        if importlib.metadata.version('playwright') < '1.15.0':
            return Connection(
                dispatcher_fiber,
                create_remote_object,
                transport)
        else:
            return Connection(
                dispatcher_fiber,
                create_remote_object,
                transport,
                loop)

    def _setup_browser(self, browser: BrowserImpl):
        version = importlib.metadata.version('playwright')
        if version < '1.17.0':
            browser._is_remote = True
            browser._is_connected_over_websocket = True
        elif version < '1.19.0':
            browser._should_close_connection_on_close = True
        else:
            browser._should_close_connection_on_close = True
            browser._local_utils = self._playwright_impl._utils

    def __enter__(self) -> SyncBrowser:
        loop: asyncio.AbstractEventLoop
        own_loop = None
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            own_loop = loop
        if loop.is_running():
            raise Error(
                """It looks like you are using Playwright Sync API inside the asyncio loop.
Please use the Async API instead."""
            )

        def greenlet_main() -> None:
            loop.run_until_complete(self._connection.run_as_sync())

            if own_loop:
                loop.run_until_complete(loop.shutdown_asyncgens())
                loop.close()

        dispatcher_fiber = greenlet(greenlet_main)
        transport = WebSocketTransport(
            loop,
            self._ws_endpoint,
            self._headers,
            self._slow_mo)
        self._connection = self._make_connection(
            dispatcher_fiber,
            create_remote_object,
            transport,
            loop)

        g_self = greenlet.getcurrent()

        def callback_wrapper(playwright_impl: PlaywrightImpl) -> None:
            self._playwright_impl = playwright_impl
            g_self.switch()

        self._connection.call_on_object_with_known_name(
            "Playwright", callback_wrapper)

        dispatcher_fiber.switch()

        pre_launched_browser = self._playwright_impl._initializer.get(
            "preLaunchedBrowser")
        assert pre_launched_browser
        browser = cast(BrowserImpl, from_channel(pre_launched_browser))
        self._setup_browser(browser)

        def handle_transport_close() -> None:
            for context in browser.contexts:
                for page in context.pages:
                    page._on_close()
                context._on_close()
            browser._on_close()

        transport.once("close", handle_transport_close)

        self._browser = SyncBrowser(browser)
        return self._browser

    def start(self) -> SyncBrowser:
        return self.__enter__()

    def __exit__(self, *args: Any) -> None:
        self._browser.close()


def connect_to_browser(
    ws_endpoint: str,
    slow_mo: float = None,
    headers: Dict[str, str] = None,
) -> SyncBrowser:
    return ConnectToBrowserContextManager(ws_endpoint, slow_mo, headers)

1# Copyright (c) Microsoft Corporation.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import asyncio
16import pathlib
17from pathlib import Path
18from typing import TYPE_CHECKING, Dict, List, Optional, Union, cast
19
20from playwright._impl._api_structures import (
21    Geolocation,
22    HttpCredentials,
23    ProxySettings,
24    ViewportSize,
25)
26from playwright._impl._api_types import Error
27from playwright._impl._browser import Browser, normalize_context_params
28from playwright._impl._browser_context import BrowserContext
29from playwright._impl._connection import (
30    ChannelOwner,
31    Connection,
32    from_channel,
33    from_nullable_channel,
34)
35from playwright._impl._helper import (
36    ColorScheme,
37    Env,
38    ForcedColors,
39    ReducedMotion,
40    locals_to_params,
41)
42from playwright._impl._transport import WebSocketTransport
43from playwright._impl._wait_helper import throw_on_timeout
44
45if TYPE_CHECKING:
46    from playwright._impl._playwright import Playwright
47
48
49class BrowserType(ChannelOwner):
50    def __init__(
51        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
52    ) -> None:
53        super().__init__(parent, type, guid, initializer)
54        _playwright: "Playwright"
55
56    def __repr__(self) -> str:
57        return f"<BrowserType name={self.name} executable_path={self.executable_path}>"
58
59    @property
60    def name(self) -> str:
61        return self._initializer["name"]
62
63    @property
64    def executable_path(self) -> str:
65        return self._initializer["executablePath"]
66
67    async def launch(
68        self,
69        executablePath: Union[str, Path] = None,
70        channel: str = None,
71        args: List[str] = None,
72        ignoreDefaultArgs: Union[bool, List[str]] = None,
73        handleSIGINT: bool = None,
74        handleSIGTERM: bool = None,
75        handleSIGHUP: bool = None,
76        timeout: float = None,
77        env: Env = None,
78        headless: bool = None,
79        devtools: bool = None,
80        proxy: ProxySettings = None,
81        downloadsPath: Union[str, Path] = None,
82        slowMo: float = None,
83        tracesDir: Union[pathlib.Path, str] = None,
84        chromiumSandbox: bool = None,
85        firefoxUserPrefs: Dict[str, Union[str, float, bool]] = None,
86    ) -> Browser:
87        params = locals_to_params(locals())
88        normalize_launch_params(params)
89        browser = cast(
90            Browser, from_channel(await self._channel.send("launch", params))
91        )
92        browser._local_utils = self._playwright._utils
93        return browser
94
95    async def launch_persistent_context(
96        self,
97        userDataDir: Union[str, Path],
98        channel: str = None,
99        executablePath: Union[str, Path] = None,
100        args: List[str] = None,
101        ignoreDefaultArgs: Union[bool, List[str]] = None,
102        handleSIGINT: bool = None,
103        handleSIGTERM: bool = None,
104        handleSIGHUP: bool = None,
105        timeout: float = None,
106        env: Env = None,
107        headless: bool = None,
108        devtools: bool = None,
109        proxy: ProxySettings = None,
110        downloadsPath: Union[str, Path] = None,
111        slowMo: float = None,
112        viewport: ViewportSize = None,
113        screen: ViewportSize = None,
114        noViewport: bool = None,
115        ignoreHTTPSErrors: bool = None,
116        javaScriptEnabled: bool = None,
117        bypassCSP: bool = None,
118        userAgent: str = None,
119        locale: str = None,
120        timezoneId: str = None,
121        geolocation: Geolocation = None,
122        permissions: List[str] = None,
123        extraHTTPHeaders: Dict[str, str] = None,
124        offline: bool = None,
125        httpCredentials: HttpCredentials = None,
126        deviceScaleFactor: float = None,
127        isMobile: bool = None,
128        hasTouch: bool = None,
129        colorScheme: ColorScheme = None,
130        reducedMotion: ReducedMotion = None,
131        forcedColors: ForcedColors = None,
132        acceptDownloads: bool = None,
133        tracesDir: Union[pathlib.Path, str] = None,
134        chromiumSandbox: bool = None,
135        recordHarPath: Union[Path, str] = None,
136        recordHarOmitContent: bool = None,
137        recordVideoDir: Union[Path, str] = None,
138        recordVideoSize: ViewportSize = None,
139        baseURL: str = None,
140        strictSelectors: bool = None,
141    ) -> BrowserContext:
142        userDataDir = str(Path(userDataDir))
143        params = locals_to_params(locals())
144        await normalize_context_params(self._connection._is_sync, params)
145        normalize_launch_params(params)
146        context = cast(
147            BrowserContext,
148            from_channel(await self._channel.send("launchPersistentContext", params)),
149        )
150        context._options = params
151        context.tracing._local_utils = self._playwright._utils
152        return context
153
154    async def connect_over_cdp(
155        self,
156        endpointURL: str,
157        timeout: float = None,
158        slow_mo: float = None,
159        headers: Dict[str, str] = None,
160    ) -> Browser:
161        params = locals_to_params(locals())
162        response = await self._channel.send_return_as_dict("connectOverCDP", params)
163        browser = cast(Browser, from_channel(response["browser"]))
164        browser._local_utils = self._playwright._utils
165
166        default_context = cast(
167            Optional[BrowserContext],
168            from_nullable_channel(response.get("defaultContext")),
169        )
170        if default_context:
171            browser._contexts.append(default_context)
172            default_context._browser = browser
173        return browser
174
175    async def connect(
176        self,
177        ws_endpoint: str,
178        timeout: float = None,
179        slow_mo: float = None,
180        headers: Dict[str, str] = None,
181    ) -> Browser:
182        if timeout is None:
183            timeout = 30000
184
185        headers = {**(headers if headers else {}), "x-playwright-browser": self.name}
186
187        transport = WebSocketTransport(
188            self._connection._loop, ws_endpoint, headers, slow_mo
189        )
190        connection = Connection(
191            self._connection._dispatcher_fiber,
192            self._connection._object_factory,
193            transport,
194            self._connection._loop,
195        )
196        connection.mark_as_remote()
197        connection._is_sync = self._connection._is_sync
198        connection._loop.create_task(connection.run())
199        playwright_future = connection.playwright_future
200
201        timeout_future = throw_on_timeout(timeout, Error("Connection timed out"))
202        done, pending = await asyncio.wait(
203            {transport.on_error_future, playwright_future, timeout_future},
204            return_when=asyncio.FIRST_COMPLETED,
205        )
206        if not playwright_future.done():
207            playwright_future.cancel()
208        if not timeout_future.done():
209            timeout_future.cancel()
210        playwright: "Playwright" = next(iter(done)).result()
211        playwright._set_selectors(self._playwright.selectors)
212        self._connection._child_ws_connections.append(connection)
213        pre_launched_browser = playwright._initializer.get("preLaunchedBrowser")
214        assert pre_launched_browser
215        browser = cast(Browser, from_channel(pre_launched_browser))
216        browser._should_close_connection_on_close = True
217        browser._local_utils = self._playwright._utils
218
219        def handle_transport_close() -> None:
220            for context in browser.contexts:
221                for page in context.pages:
222                    page._on_close()
223                context._on_close()
224            browser._on_close()
225            connection.cleanup()
226
227        transport.once("close", handle_transport_close)
228
229        return browser
230
231
232def normalize_launch_params(params: Dict) -> None:
233    if "env" in params:
234        params["env"] = [
235            {"name": name, "value": str(value)}
236            for [name, value] in params["env"].items()
237        ]
238    if "ignoreDefaultArgs" in params:
239        if params["ignoreDefaultArgs"] is True:
240            params["ignoreAllDefaultArgs"] = True
241            del params["ignoreDefaultArgs"]
242    if "executablePath" in params:
243        params["executablePath"] = str(Path(params["executablePath"]))
244    if "downloadsPath" in params:
245        params["downloadsPath"] = str(Path(params["downloadsPath"]))
246

Source: _browser_type.py

# Copyright (c) Microsoft Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import pathlib
from pathlib import Path
from typing import TYPE_CHECKING, Dict, List, Optional, Union, cast

from playwright._impl._api_structures import (
    Geolocation,
    HttpCredentials,
    ProxySettings,
    ViewportSize,
)
from playwright._impl._api_types import Error
from playwright._impl._browser import Browser, normalize_context_params
from playwright._impl._browser_context import BrowserContext
from playwright._impl._connection import (
    ChannelOwner,
    Connection,
    from_channel,
    from_nullable_channel,
)
from playwright._impl._helper import (
    ColorScheme,
    Env,
    ForcedColors,
    ReducedMotion,
    locals_to_params,
)
from playwright._impl._transport import WebSocketTransport
from playwright._impl._wait_helper import throw_on_timeout

if TYPE_CHECKING:
    from playwright._impl._playwright import Playwright


class BrowserType(ChannelOwner):
    def __init__(
        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
    ) -> None:
        super().__init__(parent, type, guid, initializer)
        _playwright: "Playwright"

    def __repr__(self) -> str:
        return f"<BrowserType name={self.name} executable_path={self.executable_path}>"

    @property
    def name(self) -> str:
        return self._initializer["name"]

    @property
    def executable_path(self) -> str:
        return self._initializer["executablePath"]

    async def launch(
        self,
        executablePath: Union[str, Path] = None,
        channel: str = None,
        args: List[str] = None,
        ignoreDefaultArgs: Union[bool, List[str]] = None,
        handleSIGINT: bool = None,
        handleSIGTERM: bool = None,
        handleSIGHUP: bool = None,
        timeout: float = None,
        env: Env = None,
        headless: bool = None,
        devtools: bool = None,
        proxy: ProxySettings = None,
        downloadsPath: Union[str, Path] = None,
        slowMo: float = None,
        tracesDir: Union[pathlib.Path, str] = None,
        chromiumSandbox: bool = None,
        firefoxUserPrefs: Dict[str, Union[str, float, bool]] = None,
    ) -> Browser:
        params = locals_to_params(locals())
        normalize_launch_params(params)
        browser = cast(
            Browser, from_channel(await self._channel.send("launch", params))
        )
        browser._local_utils = self._playwright._utils
        return browser

    async def launch_persistent_context(
        self,
        userDataDir: Union[str, Path],
        channel: str = None,
        executablePath: Union[str, Path] = None,
        args: List[str] = None,
        ignoreDefaultArgs: Union[bool, List[str]] = None,
        handleSIGINT: bool = None,
        handleSIGTERM: bool = None,
        handleSIGHUP: bool = None,
        timeout: float = None,
        env: Env = None,
        headless: bool = None,
        devtools: bool = None,
        proxy: ProxySettings = None,
        downloadsPath: Union[str, Path] = None,
        slowMo: float = None,
        viewport: ViewportSize = None,
        screen: ViewportSize = None,
        noViewport: bool = None,
        ignoreHTTPSErrors: bool = None,
        javaScriptEnabled: bool = None,
        bypassCSP: bool = None,
        userAgent: str = None,
        locale: str = None,
        timezoneId: str = None,
        geolocation: Geolocation = None,
        permissions: List[str] = None,
        extraHTTPHeaders: Dict[str, str] = None,
        offline: bool = None,
        httpCredentials: HttpCredentials = None,
        deviceScaleFactor: float = None,
        isMobile: bool = None,
        hasTouch: bool = None,
        colorScheme: ColorScheme = None,
        reducedMotion: ReducedMotion = None,
        forcedColors: ForcedColors = None,
        acceptDownloads: bool = None,
        tracesDir: Union[pathlib.Path, str] = None,
        chromiumSandbox: bool = None,
        recordHarPath: Union[Path, str] = None,
        recordHarOmitContent: bool = None,
        recordVideoDir: Union[Path, str] = None,
        recordVideoSize: ViewportSize = None,
        baseURL: str = None,
        strictSelectors: bool = None,
    ) -> BrowserContext:
        userDataDir = str(Path(userDataDir))
        params = locals_to_params(locals())
        await normalize_context_params(self._connection._is_sync, params)
        normalize_launch_params(params)
        context = cast(
            BrowserContext,
            from_channel(await self._channel.send("launchPersistentContext", params)),
        )
        context._options = params
        context.tracing._local_utils = self._playwright._utils
        return context

    async def connect_over_cdp(
        self,
        endpointURL: str,
        timeout: float = None,
        slow_mo: float = None,
        headers: Dict[str, str] = None,
    ) -> Browser:
        params = locals_to_params(locals())
        response = await self._channel.send_return_as_dict("connectOverCDP", params)
        browser = cast(Browser, from_channel(response["browser"]))
        browser._local_utils = self._playwright._utils

        default_context = cast(
            Optional[BrowserContext],
            from_nullable_channel(response.get("defaultContext")),
        )
        if default_context:
            browser._contexts.append(default_context)
            default_context._browser = browser
        return browser

    async def connect(
        self,
        ws_endpoint: str,
        timeout: float = None,
        slow_mo: float = None,
        headers: Dict[str, str] = None,
    ) -> Browser:
        if timeout is None:
            timeout = 30000

        headers = {**(headers if headers else {}), "x-playwright-browser": self.name}

        transport = WebSocketTransport(
            self._connection._loop, ws_endpoint, headers, slow_mo
        )
        connection = Connection(
            self._connection._dispatcher_fiber,
            self._connection._object_factory,
            transport,
            self._connection._loop,
        )
        connection.mark_as_remote()
        connection._is_sync = self._connection._is_sync
        connection._loop.create_task(connection.run())
        playwright_future = connection.playwright_future

        timeout_future = throw_on_timeout(timeout, Error("Connection timed out"))
        done, pending = await asyncio.wait(
            {transport.on_error_future, playwright_future, timeout_future},
            return_when=asyncio.FIRST_COMPLETED,
        )
        if not playwright_future.done():
            playwright_future.cancel()
        if not timeout_future.done():
            timeout_future.cancel()
        playwright: "Playwright" = next(iter(done)).result()
        playwright._set_selectors(self._playwright.selectors)
        self._connection._child_ws_connections.append(connection)
        pre_launched_browser = playwright._initializer.get("preLaunchedBrowser")
        assert pre_launched_browser
        browser = cast(Browser, from_channel(pre_launched_browser))
        browser._should_close_connection_on_close = True
        browser._local_utils = self._playwright._utils

        def handle_transport_close() -> None:
            for context in browser.contexts:
                for page in context.pages:
                    page._on_close()
                context._on_close()
            browser._on_close()
            connection.cleanup()

        transport.once("close", handle_transport_close)

        return browser


def normalize_launch_params(params: Dict) -> None:
    if "env" in params:
        params["env"] = [
            {"name": name, "value": str(value)}
            for [name, value] in params["env"].items()
        ]
    if "ignoreDefaultArgs" in params:
        if params["ignoreDefaultArgs"] is True:
            params["ignoreAllDefaultArgs"] = True
            del params["ignoreDefaultArgs"]
    if "executablePath" in params:
        params["executablePath"] = str(Path(params["executablePath"]))
    if "downloadsPath" in params:
        params["downloadsPath"] = str(Path(params["downloadsPath"]))