GeezerVison is showing github.com — playwright-python/playwright/async_api/_context_manager.py at main · Coder-Soder/playwright-python

Expand file tree

_context_manager.py

Copy path

More file actions

Latest commit

History

54 lines (46 loc) · 1.96 KB

_context_manager.py

Copy path

File metadata and controls

54 lines (46 loc) · 1.96 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

# 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

from typing import Any

from playwright._impl._connection import Connection

from playwright._impl._driver import compute_driver_executable

from playwright._impl._object_factory import create_remote_object

from playwright._impl._transport import PipeTransport

from playwright.async_api._generated import Playwright as AsyncPlaywright

class PlaywrightContextManager:

def __init__(self) -> None:

self._connection: Connection

async def __aenter__(self) -> AsyncPlaywright:

loop = asyncio.get_running_loop()

self._connection = Connection(

None,

create_remote_object,

PipeTransport(loop, compute_driver_executable()),

loop,

)

loop.create_task(self._connection.run())

playwright_future = self._connection.playwright_future

done, pending = await asyncio.wait(

{self._connection._transport.on_error_future, playwright_future},

return_when=asyncio.FIRST_COMPLETED,

)

if not playwright_future.done():

playwright_future.cancel()

playwright = AsyncPlaywright(next(iter(done)).result())

playwright.stop = self.__aexit__ # type: ignore

return playwright

async def start(self) -> AsyncPlaywright:

return await self.__aenter__()

async def __aexit__(self, *args: Any) -> None:

await self._connection.stop_async()