325 lines
11 KiB
Python
325 lines
11 KiB
Python
|
"""
|
||
|
|
This type stub file was generated by pyright.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from typing import Awaitable, Dict, List, Optional, TYPE_CHECKING, Union
|
||
|
|
from pyee import EventEmitter
|
||
|
|
from pyppeteer.connection import CDPSession
|
||
|
|
from pyppeteer.frame_manager import Frame, FrameManager
|
||
|
|
|
||
|
|
"""Network Manager module."""
|
||
|
|
if TYPE_CHECKING:
|
||
|
|
...
|
||
|
|
logger = ...
|
||
|
|
class NetworkManager(EventEmitter):
|
||
|
|
"""NetworkManager class."""
|
||
|
|
Events = ...
|
||
|
|
def __init__(self, client: CDPSession, frameManager: FrameManager) -> None:
|
||
|
|
"""Make new NetworkManager."""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def authenticate(self, credentials: Dict[str, str]) -> None:
|
||
|
|
"""Provide credentials for http auth."""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def setExtraHTTPHeaders(self, extraHTTPHeaders: Dict[str, str]) -> None:
|
||
|
|
"""Set extra http headers."""
|
||
|
|
...
|
||
|
|
|
||
|
|
def extraHTTPHeaders(self) -> Dict[str, str]:
|
||
|
|
"""Get extra http headers."""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def setOfflineMode(self, value: bool) -> None:
|
||
|
|
"""Change offline mode enable/disable."""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def setUserAgent(self, userAgent: str) -> None:
|
||
|
|
"""Set user agent."""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def setRequestInterception(self, value: bool) -> None:
|
||
|
|
"""Enable request interception."""
|
||
|
|
...
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
class Request:
|
||
|
|
"""Request class.
|
||
|
|
|
||
|
|
Whenever the page sends a request, such as for a network resource, the
|
||
|
|
following events are emitted by pyppeteer's page:
|
||
|
|
|
||
|
|
- ``'request'``: emitted when the request is issued by the page.
|
||
|
|
- ``'response'``: emitted when/if the response is received for the request.
|
||
|
|
- ``'requestfinished'``: emitted when the response body is downloaded and
|
||
|
|
the request is complete.
|
||
|
|
|
||
|
|
If request fails at some point, then instead of ``'requestfinished'`` event
|
||
|
|
(and possibly instead of ``'response'`` event), the ``'requestfailed'``
|
||
|
|
event is emitted.
|
||
|
|
|
||
|
|
If request gets a ``'redirect'`` response, the request is successfully
|
||
|
|
finished with the ``'requestfinished'`` event, and a new request is issued
|
||
|
|
to a redirect url.
|
||
|
|
"""
|
||
|
|
def __init__(self, client: CDPSession, requestId: Optional[str], interceptionId: Optional[str], isNavigationRequest: bool, allowInterception: bool, url: str, resourceType: str, payload: dict, frame: Optional[Frame], redirectChain: List[Request]) -> None:
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def url(self) -> str:
|
||
|
|
"""URL of this request."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def resourceType(self) -> str:
|
||
|
|
"""Resource type of this request perceived by the rendering engine.
|
||
|
|
|
||
|
|
ResourceType will be one of the following: ``document``,
|
||
|
|
``stylesheet``, ``image``, ``media``, ``font``, ``script``,
|
||
|
|
``texttrack``, ``xhr``, ``fetch``, ``eventsource``, ``websocket``,
|
||
|
|
``manifest``, ``other``.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def method(self) -> Optional[str]:
|
||
|
|
"""Return this request's method (GET, POST, etc.)."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def postData(self) -> Optional[str]:
|
||
|
|
"""Return post body of this request."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def headers(self) -> Dict:
|
||
|
|
"""Return a dictionary of HTTP headers of this request.
|
||
|
|
|
||
|
|
All header names are lower-case.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def response(self) -> Optional[Response]:
|
||
|
|
"""Return matching :class:`Response` object, or ``None``.
|
||
|
|
|
||
|
|
If the response has not been received, return ``None``.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def frame(self) -> Optional[Frame]:
|
||
|
|
"""Return a matching :class:`~pyppeteer.frame_manager.frame` object.
|
||
|
|
|
||
|
|
Return ``None`` if navigating to error page.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
def isNavigationRequest(self) -> bool:
|
||
|
|
"""Whether this request is driving frame's navigation."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def redirectChain(self) -> List[Request]:
|
||
|
|
"""Return chain of requests initiated to fetch a resource.
|
||
|
|
|
||
|
|
* If there are no redirects and request was successful, the chain will
|
||
|
|
be empty.
|
||
|
|
* If a server responds with at least a single redirect, then the chain
|
||
|
|
will contain all the requests that were redirected.
|
||
|
|
|
||
|
|
``redirectChain`` is shared between all the requests of the same chain.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
def failure(self) -> Optional[Dict]:
|
||
|
|
"""Return error text.
|
||
|
|
|
||
|
|
Return ``None`` unless this request was failed, as reported by
|
||
|
|
``requestfailed`` event.
|
||
|
|
|
||
|
|
When request failed, this method return dictionary which has a
|
||
|
|
``errorText`` field, which contains human-readable error message, e.g.
|
||
|
|
``'net::ERR_RAILED'``.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def continue_(self, overrides: Dict = ...) -> None:
|
||
|
|
"""Continue request with optional request overrides.
|
||
|
|
|
||
|
|
To use this method, request interception should be enabled by
|
||
|
|
:meth:`pyppeteer.page.Page.setRequestInterception`. If request
|
||
|
|
interception is not enabled, raise ``NetworkError``.
|
||
|
|
|
||
|
|
``overrides`` can have the following fields:
|
||
|
|
|
||
|
|
* ``url`` (str): If set, the request url will be changed.
|
||
|
|
* ``method`` (str): If set, change the request method (e.g. ``GET``).
|
||
|
|
* ``postData`` (str): If set, change the post data or request.
|
||
|
|
* ``headers`` (dict): If set, change the request HTTP header.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def respond(self, response: Dict) -> None:
|
||
|
|
"""Fulfills request with given response.
|
||
|
|
|
||
|
|
To use this, request interception should by enabled by
|
||
|
|
:meth:`pyppeteer.page.Page.setRequestInterception`. Request
|
||
|
|
interception is not enabled, raise ``NetworkError``.
|
||
|
|
|
||
|
|
``response`` is a dictionary which can have the following fields:
|
||
|
|
|
||
|
|
* ``status`` (int): Response status code, defaults to 200.
|
||
|
|
* ``headers`` (dict): Optional response headers.
|
||
|
|
* ``contentType`` (str): If set, equals to setting ``Content-Type``
|
||
|
|
response header.
|
||
|
|
* ``body`` (str|bytes): Optional response body.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def abort(self, errorCode: str = ...) -> None:
|
||
|
|
"""Abort request.
|
||
|
|
|
||
|
|
To use this, request interception should be enabled by
|
||
|
|
:meth:`pyppeteer.page.Page.setRequestInterception`.
|
||
|
|
If request interception is not enabled, raise ``NetworkError``.
|
||
|
|
|
||
|
|
``errorCode`` is an optional error code string. Defaults to ``failed``,
|
||
|
|
could be one of the following:
|
||
|
|
|
||
|
|
- ``aborted``: An operation was aborted (due to user action).
|
||
|
|
- ``accessdenied``: Permission to access a resource, other than the
|
||
|
|
network, was denied.
|
||
|
|
- ``addressunreachable``: The IP address is unreachable. This usually
|
||
|
|
means that there is no route to the specified host or network.
|
||
|
|
- ``blockedbyclient``: The client chose to block the request.
|
||
|
|
- ``blockedbyresponse``: The request failed because the request was
|
||
|
|
delivered along with requirements which are not met
|
||
|
|
('X-Frame-Options' and 'Content-Security-Policy' ancestor check,
|
||
|
|
for instance).
|
||
|
|
- ``connectionaborted``: A connection timeout as a result of not
|
||
|
|
receiving an ACK for data sent.
|
||
|
|
- ``connectionclosed``: A connection was closed (corresponding to a TCP
|
||
|
|
FIN).
|
||
|
|
- ``connectionfailed``: A connection attempt failed.
|
||
|
|
- ``connectionrefused``: A connection attempt was refused.
|
||
|
|
- ``connectionreset``: A connection was reset (corresponding to a TCP
|
||
|
|
RST).
|
||
|
|
- ``internetdisconnected``: The Internet connection has been lost.
|
||
|
|
- ``namenotresolved``: The host name could not be resolved.
|
||
|
|
- ``timedout``: An operation timed out.
|
||
|
|
- ``failed``: A generic failure occurred.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
errorReasons = ...
|
||
|
|
class Response:
|
||
|
|
"""Response class represents responses which are received by ``Page``."""
|
||
|
|
def __init__(self, client: CDPSession, request: Request, status: int, headers: Dict[str, str], fromDiskCache: bool, fromServiceWorker: bool, securityDetails: Dict = ...) -> None:
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def url(self) -> str:
|
||
|
|
"""URL of the response."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def ok(self) -> bool:
|
||
|
|
"""Return bool whether this request is successful (200-299) or not."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def status(self) -> int:
|
||
|
|
"""Status code of the response."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def headers(self) -> Dict:
|
||
|
|
"""Return dictionary of HTTP headers of this response.
|
||
|
|
|
||
|
|
All header names are lower-case.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def securityDetails(self) -> Union[Dict, SecurityDetails]:
|
||
|
|
"""Return security details associated with this response.
|
||
|
|
|
||
|
|
Security details if the response was received over the secure
|
||
|
|
connection, or `None` otherwise.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
def buffer(self) -> Awaitable[bytes]:
|
||
|
|
"""Return awaitable which resolves to bytes with response body."""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def text(self) -> str:
|
||
|
|
"""Get text representation of response body."""
|
||
|
|
...
|
||
|
|
|
||
|
|
async def json(self) -> dict:
|
||
|
|
"""Get JSON representation of response body."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def request(self) -> Request:
|
||
|
|
"""Get matching :class:`Request` object."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def fromCache(self) -> bool:
|
||
|
|
"""Return ``True`` if the response was served from cache.
|
||
|
|
|
||
|
|
Here `cache` is either the browser's disk cache or memory cache.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def fromServiceWorker(self) -> bool:
|
||
|
|
"""Return ``True`` if the response was served by a service worker."""
|
||
|
|
...
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def generateRequestHash(request: dict) -> str:
|
||
|
|
"""Generate request hash."""
|
||
|
|
...
|
||
|
|
|
||
|
|
class SecurityDetails:
|
||
|
|
"""Class represents responses which are received by page."""
|
||
|
|
def __init__(self, subjectName: str, issuer: str, validFrom: int, validTo: int, protocol: str) -> None:
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def subjectName(self) -> str:
|
||
|
|
"""Return the subject to which the certificate was issued to."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def issuer(self) -> str:
|
||
|
|
"""Return a string with the name of issuer of the certificate."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def validFrom(self) -> int:
|
||
|
|
"""Return UnixTime of the start of validity of the certificate."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def validTo(self) -> int:
|
||
|
|
"""Return UnixTime of the end of validity of the certificate."""
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
def protocol(self) -> str:
|
||
|
|
"""Return string of with the security protocol, e.g. "TLS1.2"."""
|
||
|
|
...
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
statusTexts = ...
|