jpayne@7: from __future__ import annotations
jpayne@7: 
jpayne@7: import typing
jpayne@7: 
jpayne@7: from .util.connection import _TYPE_SOCKET_OPTIONS
jpayne@7: from .util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT
jpayne@7: from .util.url import Url
jpayne@7: 
jpayne@7: _TYPE_BODY = typing.Union[bytes, typing.IO[typing.Any], typing.Iterable[bytes], str]
jpayne@7: 
jpayne@7: 
jpayne@7: class ProxyConfig(typing.NamedTuple):
jpayne@7:     ssl_context: ssl.SSLContext | None
jpayne@7:     use_forwarding_for_https: bool
jpayne@7:     assert_hostname: None | str | Literal[False]
jpayne@7:     assert_fingerprint: str | None
jpayne@7: 
jpayne@7: 
jpayne@7: class _ResponseOptions(typing.NamedTuple):
jpayne@7:     # TODO: Remove this in favor of a better
jpayne@7:     # HTTP request/response lifecycle tracking.
jpayne@7:     request_method: str
jpayne@7:     request_url: str
jpayne@7:     preload_content: bool
jpayne@7:     decode_content: bool
jpayne@7:     enforce_content_length: bool
jpayne@7: 
jpayne@7: 
jpayne@7: if typing.TYPE_CHECKING:
jpayne@7:     import ssl
jpayne@7:     from typing import Literal, Protocol
jpayne@7: 
jpayne@7:     from .response import BaseHTTPResponse
jpayne@7: 
jpayne@7:     class BaseHTTPConnection(Protocol):
jpayne@7:         default_port: typing.ClassVar[int]
jpayne@7:         default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS]
jpayne@7: 
jpayne@7:         host: str
jpayne@7:         port: int
jpayne@7:         timeout: None | (
jpayne@7:             float
jpayne@7:         )  # Instance doesn't store _DEFAULT_TIMEOUT, must be resolved.
jpayne@7:         blocksize: int
jpayne@7:         source_address: tuple[str, int] | None
jpayne@7:         socket_options: _TYPE_SOCKET_OPTIONS | None
jpayne@7: 
jpayne@7:         proxy: Url | None
jpayne@7:         proxy_config: ProxyConfig | None
jpayne@7: 
jpayne@7:         is_verified: bool
jpayne@7:         proxy_is_verified: bool | None
jpayne@7: 
jpayne@7:         def __init__(
jpayne@7:             self,
jpayne@7:             host: str,
jpayne@7:             port: int | None = None,
jpayne@7:             *,
jpayne@7:             timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
jpayne@7:             source_address: tuple[str, int] | None = None,
jpayne@7:             blocksize: int = 8192,
jpayne@7:             socket_options: _TYPE_SOCKET_OPTIONS | None = ...,
jpayne@7:             proxy: Url | None = None,
jpayne@7:             proxy_config: ProxyConfig | None = None,
jpayne@7:         ) -> None:
jpayne@7:             ...
jpayne@7: 
jpayne@7:         def set_tunnel(
jpayne@7:             self,
jpayne@7:             host: str,
jpayne@7:             port: int | None = None,
jpayne@7:             headers: typing.Mapping[str, str] | None = None,
jpayne@7:             scheme: str = "http",
jpayne@7:         ) -> None:
jpayne@7:             ...
jpayne@7: 
jpayne@7:         def connect(self) -> None:
jpayne@7:             ...
jpayne@7: 
jpayne@7:         def request(
jpayne@7:             self,
jpayne@7:             method: str,
jpayne@7:             url: str,
jpayne@7:             body: _TYPE_BODY | None = None,
jpayne@7:             headers: typing.Mapping[str, str] | None = None,
jpayne@7:             # We know *at least* botocore is depending on the order of the
jpayne@7:             # first 3 parameters so to be safe we only mark the later ones
jpayne@7:             # as keyword-only to ensure we have space to extend.
jpayne@7:             *,
jpayne@7:             chunked: bool = False,
jpayne@7:             preload_content: bool = True,
jpayne@7:             decode_content: bool = True,
jpayne@7:             enforce_content_length: bool = True,
jpayne@7:         ) -> None:
jpayne@7:             ...
jpayne@7: 
jpayne@7:         def getresponse(self) -> BaseHTTPResponse:
jpayne@7:             ...
jpayne@7: 
jpayne@7:         def close(self) -> None:
jpayne@7:             ...
jpayne@7: 
jpayne@7:         @property
jpayne@7:         def is_closed(self) -> bool:
jpayne@7:             """Whether the connection either is brand new or has been previously closed.
jpayne@7:             If this property is True then both ``is_connected`` and ``has_connected_to_proxy``
jpayne@7:             properties must be False.
jpayne@7:             """
jpayne@7: 
jpayne@7:         @property
jpayne@7:         def is_connected(self) -> bool:
jpayne@7:             """Whether the connection is actively connected to any origin (proxy or target)"""
jpayne@7: 
jpayne@7:         @property
jpayne@7:         def has_connected_to_proxy(self) -> bool:
jpayne@7:             """Whether the connection has successfully connected to its proxy.
jpayne@7:             This returns False if no proxy is in use. Used to determine whether
jpayne@7:             errors are coming from the proxy layer or from tunnelling to the target origin.
jpayne@7:             """
jpayne@7: 
jpayne@7:     class BaseHTTPSConnection(BaseHTTPConnection, Protocol):
jpayne@7:         default_port: typing.ClassVar[int]
jpayne@7:         default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS]
jpayne@7: 
jpayne@7:         # Certificate verification methods
jpayne@7:         cert_reqs: int | str | None
jpayne@7:         assert_hostname: None | str | Literal[False]
jpayne@7:         assert_fingerprint: str | None
jpayne@7:         ssl_context: ssl.SSLContext | None
jpayne@7: 
jpayne@7:         # Trusted CAs
jpayne@7:         ca_certs: str | None
jpayne@7:         ca_cert_dir: str | None
jpayne@7:         ca_cert_data: None | str | bytes
jpayne@7: 
jpayne@7:         # TLS version
jpayne@7:         ssl_minimum_version: int | None
jpayne@7:         ssl_maximum_version: int | None
jpayne@7:         ssl_version: int | str | None  # Deprecated
jpayne@7: 
jpayne@7:         # Client certificates
jpayne@7:         cert_file: str | None
jpayne@7:         key_file: str | None
jpayne@7:         key_password: str | None
jpayne@7: 
jpayne@7:         def __init__(
jpayne@7:             self,
jpayne@7:             host: str,
jpayne@7:             port: int | None = None,
jpayne@7:             *,
jpayne@7:             timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
jpayne@7:             source_address: tuple[str, int] | None = None,
jpayne@7:             blocksize: int = 16384,
jpayne@7:             socket_options: _TYPE_SOCKET_OPTIONS | None = ...,
jpayne@7:             proxy: Url | None = None,
jpayne@7:             proxy_config: ProxyConfig | None = None,
jpayne@7:             cert_reqs: int | str | None = None,
jpayne@7:             assert_hostname: None | str | Literal[False] = None,
jpayne@7:             assert_fingerprint: str | None = None,
jpayne@7:             server_hostname: str | None = None,
jpayne@7:             ssl_context: ssl.SSLContext | None = None,
jpayne@7:             ca_certs: str | None = None,
jpayne@7:             ca_cert_dir: str | None = None,
jpayne@7:             ca_cert_data: None | str | bytes = None,
jpayne@7:             ssl_minimum_version: int | None = None,
jpayne@7:             ssl_maximum_version: int | None = None,
jpayne@7:             ssl_version: int | str | None = None,  # Deprecated
jpayne@7:             cert_file: str | None = None,
jpayne@7:             key_file: str | None = None,
jpayne@7:             key_password: str | None = None,
jpayne@7:         ) -> None:
jpayne@7:             ...