D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
hc_python
/
lib64
/
python3.8
/
site-packages
/
urllib3
/
contrib
/
emscripten
/
Filename :
connection.py
back
Copy
from __future__ import annotations import os import typing # use http.client.HTTPException for consistency with non-emscripten from http.client import HTTPException as HTTPException # noqa: F401 from http.client import ResponseNotReady from ..._base_connection import _TYPE_BODY from ...connection import HTTPConnection, ProxyConfig, port_by_scheme from ...exceptions import TimeoutError from ...response import BaseHTTPResponse from ...util.connection import _TYPE_SOCKET_OPTIONS from ...util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT from ...util.url import Url from .fetch import _RequestError, _TimeoutError, send_request, send_streaming_request from .request import EmscriptenRequest from .response import EmscriptenHttpResponseWrapper, EmscriptenResponse if typing.TYPE_CHECKING: from ..._base_connection import BaseHTTPConnection, BaseHTTPSConnection class EmscriptenHTTPConnection: default_port: typing.ClassVar[int] = port_by_scheme["http"] default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS] timeout: None | (float) host: str port: int blocksize: int source_address: tuple[str, int] | None socket_options: _TYPE_SOCKET_OPTIONS | None proxy: Url | None proxy_config: ProxyConfig | None is_verified: bool = False proxy_is_verified: bool | None = None _response: EmscriptenResponse | None def __init__( self, host: str, port: int = 0, *, timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, source_address: tuple[str, int] | None = None, blocksize: int = 8192, socket_options: _TYPE_SOCKET_OPTIONS | None = None, proxy: Url | None = None, proxy_config: ProxyConfig | None = None, ) -> None: self.host = host self.port = port self.timeout = timeout if isinstance(timeout, float) else 0.0 self.scheme = "http" self._closed = True self._response = None # ignore these things because we don't # have control over that stuff self.proxy = None self.proxy_config = None self.blocksize = blocksize self.source_address = None self.socket_options = None self.is_verified = False def set_tunnel( self, host: str, port: int | None = 0, headers: typing.Mapping[str, str] | None = None, scheme: str = "http", ) -> None: pass def connect(self) -> None: pass def request( self, method: str, url: str, body: _TYPE_BODY | None = None, headers: typing.Mapping[str, str] | None = None, # We know *at least* botocore is depending on the order of the # first 3 parameters so to be safe we only mark the later ones # as keyword-only to ensure we have space to extend. *, chunked: bool = False, preload_content: bool = True, decode_content: bool = True, enforce_content_length: bool = True, ) -> None: self._closed = False if url.startswith("/"): # no scheme / host / port included, make a full url url = f"{self.scheme}://{self.host}:{self.port}" + url request = EmscriptenRequest( url=url, method=method, timeout=self.timeout if self.timeout else 0, decode_content=decode_content, ) request.set_body(body) if headers: for k, v in headers.items(): request.set_header(k, v) self._response = None try: if not preload_content: self._response = send_streaming_request(request) if self._response is None: self._response = send_request(request) except _TimeoutError as e: raise TimeoutError(e.message) from e except _RequestError as e: raise HTTPException(e.message) from e def getresponse(self) -> BaseHTTPResponse: if self._response is not None: return EmscriptenHttpResponseWrapper( internal_response=self._response, url=self._response.request.url, connection=self, ) else: raise ResponseNotReady() def close(self) -> None: self._closed = True self._response = None @property def is_closed(self) -> bool: """Whether the connection either is brand new or has been previously closed. If this property is True then both ``is_connected`` and ``has_connected_to_proxy`` properties must be False. """ return self._closed @property def is_connected(self) -> bool: """Whether the connection is actively connected to any origin (proxy or target)""" return True @property def has_connected_to_proxy(self) -> bool: """Whether the connection has successfully connected to its proxy. This returns False if no proxy is in use. Used to determine whether errors are coming from the proxy layer or from tunnelling to the target origin. """ return False class EmscriptenHTTPSConnection(EmscriptenHTTPConnection): default_port = port_by_scheme["https"] # all this is basically ignored, as browser handles https cert_reqs: int | str | None = None ca_certs: str | None = None ca_cert_dir: str | None = None ca_cert_data: None | str | bytes = None cert_file: str | None key_file: str | None key_password: str | None ssl_context: typing.Any | None ssl_version: int | str | None = None ssl_minimum_version: int | None = None ssl_maximum_version: int | None = None assert_hostname: None | str | typing.Literal[False] assert_fingerprint: str | None = None def __init__( self, host: str, port: int = 0, *, timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, source_address: tuple[str, int] | None = None, blocksize: int = 16384, socket_options: None | _TYPE_SOCKET_OPTIONS = HTTPConnection.default_socket_options, proxy: Url | None = None, proxy_config: ProxyConfig | None = None, cert_reqs: int | str | None = None, assert_hostname: None | str | typing.Literal[False] = None, assert_fingerprint: str | None = None, server_hostname: str | None = None, ssl_context: typing.Any | None = None, ca_certs: str | None = None, ca_cert_dir: str | None = None, ca_cert_data: None | str | bytes = None, ssl_minimum_version: int | None = None, ssl_maximum_version: int | None = None, ssl_version: int | str | None = None, # Deprecated cert_file: str | None = None, key_file: str | None = None, key_password: str | None = None, ) -> None: super().__init__( host, port=port, timeout=timeout, source_address=source_address, blocksize=blocksize, socket_options=socket_options, proxy=proxy, proxy_config=proxy_config, ) self.scheme = "https" self.key_file = key_file self.cert_file = cert_file self.key_password = key_password self.ssl_context = ssl_context self.server_hostname = server_hostname self.assert_hostname = assert_hostname self.assert_fingerprint = assert_fingerprint self.ssl_version = ssl_version self.ssl_minimum_version = ssl_minimum_version self.ssl_maximum_version = ssl_maximum_version self.ca_certs = ca_certs and os.path.expanduser(ca_certs) self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir) self.ca_cert_data = ca_cert_data self.cert_reqs = None # The browser will automatically verify all requests. # We have no control over that setting. self.is_verified = True def set_cert( self, key_file: str | None = None, cert_file: str | None = None, cert_reqs: int | str | None = None, key_password: str | None = None, ca_certs: str | None = None, assert_hostname: None | str | typing.Literal[False] = None, assert_fingerprint: str | None = None, ca_cert_dir: str | None = None, ca_cert_data: None | str | bytes = None, ) -> None: pass # verify that this class implements BaseHTTP(s) connection correctly if typing.TYPE_CHECKING: _supports_http_protocol: BaseHTTPConnection = EmscriptenHTTPConnection("", 0) _supports_https_protocol: BaseHTTPSConnection = EmscriptenHTTPSConnection("", 0)