Source code for pyppeteer.target
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Target module."""
import asyncio
from typing import Any, Callable, Coroutine, Dict, List, Optional
from typing import TYPE_CHECKING
from pyppeteer.connection import CDPSession
from pyppeteer.page import Page
if TYPE_CHECKING:
from pyppeteer.browser import Browser, BrowserContext # noqa: F401
[docs]class Target(object):
"""Browser's target class."""
def __init__(self, targetInfo: Dict, browserContext: 'BrowserContext',
sessionFactory: Callable[[], Coroutine[Any, Any, CDPSession]],
ignoreHTTPSErrors: bool, setDefaultViewport: bool,
screenshotTaskQueue: List, loop: asyncio.AbstractEventLoop
) -> None:
self._targetInfo = targetInfo
self._browserContext = browserContext
self._targetId = targetInfo.get('targetId', '')
self._sessionFactory = sessionFactory
self._ignoreHTTPSErrors = ignoreHTTPSErrors
self._setDefaultViewport = setDefaultViewport
self._screenshotTaskQueue = screenshotTaskQueue
self._loop = loop
self._page: Optional[Page] = None
self._initializedPromise = self._loop.create_future()
self._isClosedPromise = self._loop.create_future()
self._isInitialized = (self._targetInfo['type'] != 'page'
or self._targetInfo['url'] != '')
if self._isInitialized:
self._initializedCallback(True)
def _initializedCallback(self, bl: bool) -> None:
# TODO: this may cause error on page close
if self._initializedPromise.done():
self._initializedPromise = self._loop.create_future()
self._initializedPromise.set_result(bl)
def _closedCallback(self) -> None:
self._isClosedPromise.set_result(None)
[docs] async def createCDPSession(self) -> CDPSession:
"""Create a Chrome Devtools Protocol session attached to the target."""
return await self._sessionFactory()
[docs] async def page(self) -> Optional[Page]:
"""Get page of this target.
If the target is not of type "page" or "background_page", return
``None``.
"""
if (self._targetInfo['type'] in ['page', 'background_page'] and
self._page is None):
client = await self._sessionFactory()
new_page = await Page.create(
client, self,
self._ignoreHTTPSErrors,
self._setDefaultViewport,
self._screenshotTaskQueue,
)
self._page = new_page
return new_page
return self._page
@property
def url(self) -> str:
"""Get url of this target."""
return self._targetInfo['url']
@property
def type(self) -> str:
"""Get type of this target.
Type can be ``'page'``, ``'background_page'``, ``'service_worker'``,
``'browser'``, or ``'other'``.
"""
_type = self._targetInfo['type']
if _type in ['page', 'background_page', 'service_worker', 'browser']:
return _type
return 'other'
@property
def browser(self) -> 'Browser':
"""Get the browser the target belongs to."""
return self._browserContext.browser
@property
def browserContext(self) -> 'BrowserContext':
"""Return the browser context the target belongs to."""
return self._browserContext
@property
def opener(self) -> Optional['Target']:
"""Get the target that opened this target.
Top-level targets return ``None``.
"""
openerId = self._targetInfo.get('openerId')
if openerId is None:
return None
return self.browser._targets.get(openerId)
def _targetInfoChanged(self, targetInfo: Dict) -> None:
self._targetInfo = targetInfo
if not self._isInitialized and (self._targetInfo['type'] != 'page' or
self._targetInfo['url'] != ''):
self._isInitialized = True
self._initializedCallback(True)
return