codekingpro/portable-devtools

1from __future__ import annotations2 3import typing4 5from .util.connection import _TYPE_SOCKET_OPTIONS6from .util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT7from .util.url import Url8 9_TYPE_BODY = typing.Union[10    bytes, typing.IO[typing.Any], typing.Iterable[bytes | str], str11]12 13 14class ProxyConfig(typing.NamedTuple):15    ssl_context: ssl.SSLContext | None16    use_forwarding_for_https: bool17    assert_hostname: None | str | typing.Literal[False]18    assert_fingerprint: str | None19 20 21class _ResponseOptions(typing.NamedTuple):22    # TODO: Remove this in favor of a better23    # HTTP request/response lifecycle tracking.24    request_method: str25    request_url: str26    preload_content: bool27    decode_content: bool28    enforce_content_length: bool29 30 31if typing.TYPE_CHECKING:32    import ssl33    from typing import Protocol34 35    from .response import BaseHTTPResponse36 37    class BaseHTTPConnection(Protocol):38        default_port: typing.ClassVar[int]39        default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS]40 41        host: str42        port: int43        timeout: None | (44            float45        )  # Instance doesn't store _DEFAULT_TIMEOUT, must be resolved.46        blocksize: int47        source_address: tuple[str, int] | None48        socket_options: _TYPE_SOCKET_OPTIONS | None49 50        proxy: Url | None51        proxy_config: ProxyConfig | None52 53        is_verified: bool54        proxy_is_verified: bool | None55 56        def __init__(57            self,58            host: str,59            port: int | None = None,60            *,61            timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,62            source_address: tuple[str, int] | None = None,63            blocksize: int = 8192,64            socket_options: _TYPE_SOCKET_OPTIONS | None = ...,65            proxy: Url | None = None,66            proxy_config: ProxyConfig | None = None,67        ) -> None: ...68 69        def set_tunnel(70            self,71            host: str,72            port: int | None = None,73            headers: typing.Mapping[str, str] | None = None,74            scheme: str = "http",75        ) -> None: ...76 77        def connect(self) -> None: ...78 79        def request(80            self,81            method: str,82            url: str,83            body: _TYPE_BODY | None = None,84            headers: typing.Mapping[str, str] | None = None,85            # We know *at least* botocore is depending on the order of the86            # first 3 parameters so to be safe we only mark the later ones87            # as keyword-only to ensure we have space to extend.88            *,89            chunked: bool = False,90            preload_content: bool = True,91            decode_content: bool = True,92            enforce_content_length: bool = True,93        ) -> None: ...94 95        def getresponse(self) -> BaseHTTPResponse: ...96 97        def close(self) -> None: ...98 99        @property100        def is_closed(self) -> bool:101            """Whether the connection either is brand new or has been previously closed.102            If this property is True then both ``is_connected`` and ``has_connected_to_proxy``103            properties must be False.104            """105 106        @property107        def is_connected(self) -> bool:108            """Whether the connection is actively connected to any origin (proxy or target)"""109 110        @property111        def has_connected_to_proxy(self) -> bool:112            """Whether the connection has successfully connected to its proxy.113            This returns False if no proxy is in use. Used to determine whether114            errors are coming from the proxy layer or from tunnelling to the target origin.115            """116 117    class BaseHTTPSConnection(BaseHTTPConnection, Protocol):118        default_port: typing.ClassVar[int]119        default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS]120 121        # Certificate verification methods122        cert_reqs: int | str | None123        assert_hostname: None | str | typing.Literal[False]124        assert_fingerprint: str | None125        ssl_context: ssl.SSLContext | None126 127        # Trusted CAs128        ca_certs: str | None129        ca_cert_dir: str | None130        ca_cert_data: None | str | bytes131 132        # TLS version133        ssl_minimum_version: int | None134        ssl_maximum_version: int | None135        ssl_version: int | str | None  # Deprecated136 137        # Client certificates138        cert_file: str | None139        key_file: str | None140        key_password: str | None141 142        def __init__(143            self,144            host: str,145            port: int | None = None,146            *,147            timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,148            source_address: tuple[str, int] | None = None,149            blocksize: int = 16384,150            socket_options: _TYPE_SOCKET_OPTIONS | None = ...,151            proxy: Url | None = None,152            proxy_config: ProxyConfig | None = None,153            cert_reqs: int | str | None = None,154            assert_hostname: None | str | typing.Literal[False] = None,155            assert_fingerprint: str | None = None,156            server_hostname: str | None = None,157            ssl_context: ssl.SSLContext | None = None,158            ca_certs: str | None = None,159            ca_cert_dir: str | None = None,160            ca_cert_data: None | str | bytes = None,161            ssl_minimum_version: int | None = None,162            ssl_maximum_version: int | None = None,163            ssl_version: int | str | None = None,  # Deprecated164            cert_file: str | None = None,165            key_file: str | None = None,166            key_password: str | None = None,167        ) -> None: ...168