codekingpro/portable-devtools

1from __future__ import annotations2 3import os4import typing5 6from ._models import Headers7from ._types import CertTypes, HeaderTypes, TimeoutTypes8from ._urls import URL9 10if typing.TYPE_CHECKING:11    import ssl  # pragma: no cover12 13__all__ = ["Limits", "Proxy", "Timeout", "create_ssl_context"]14 15 16class UnsetType:17    pass  # pragma: no cover18 19 20UNSET = UnsetType()21 22 23def create_ssl_context(24    verify: ssl.SSLContext | str | bool = True,25    cert: CertTypes | None = None,26    trust_env: bool = True,27) -> ssl.SSLContext:28    import ssl29    import warnings30 31    import certifi32 33    if verify is True:34        if trust_env and os.environ.get("SSL_CERT_FILE"):  # pragma: nocover35            ctx = ssl.create_default_context(cafile=os.environ["SSL_CERT_FILE"])36        elif trust_env and os.environ.get("SSL_CERT_DIR"):  # pragma: nocover37            ctx = ssl.create_default_context(capath=os.environ["SSL_CERT_DIR"])38        else:39            # Default case...40            ctx = ssl.create_default_context(cafile=certifi.where())41    elif verify is False:42        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)43        ctx.check_hostname = False44        ctx.verify_mode = ssl.CERT_NONE45    elif isinstance(verify, str):  # pragma: nocover46        message = (47            "`verify=<str>` is deprecated. "48            "Use `verify=ssl.create_default_context(cafile=...)` "49            "or `verify=ssl.create_default_context(capath=...)` instead."50        )51        warnings.warn(message, DeprecationWarning)52        if os.path.isdir(verify):53            return ssl.create_default_context(capath=verify)54        return ssl.create_default_context(cafile=verify)55    else:56        ctx = verify57 58    if cert:  # pragma: nocover59        message = (60            "`cert=...` is deprecated. Use `verify=<ssl_context>` instead,"61            "with `.load_cert_chain()` to configure the certificate chain."62        )63        warnings.warn(message, DeprecationWarning)64        if isinstance(cert, str):65            ctx.load_cert_chain(cert)66        else:67            ctx.load_cert_chain(*cert)68 69    return ctx70 71 72class Timeout:73    """74    Timeout configuration.75 76    **Usage**:77 78    Timeout(None)               # No timeouts.79    Timeout(5.0)                # 5s timeout on all operations.80    Timeout(None, connect=5.0)  # 5s timeout on connect, no other timeouts.81    Timeout(5.0, connect=10.0)  # 10s timeout on connect. 5s timeout elsewhere.82    Timeout(5.0, pool=None)     # No timeout on acquiring connection from pool.83                                # 5s timeout elsewhere.84    """85 86    def __init__(87        self,88        timeout: TimeoutTypes | UnsetType = UNSET,89        *,90        connect: None | float | UnsetType = UNSET,91        read: None | float | UnsetType = UNSET,92        write: None | float | UnsetType = UNSET,93        pool: None | float | UnsetType = UNSET,94    ) -> None:95        if isinstance(timeout, Timeout):96            # Passed as a single explicit Timeout.97            assert connect is UNSET98            assert read is UNSET99            assert write is UNSET100            assert pool is UNSET101            self.connect = timeout.connect  # type: typing.Optional[float]102            self.read = timeout.read  # type: typing.Optional[float]103            self.write = timeout.write  # type: typing.Optional[float]104            self.pool = timeout.pool  # type: typing.Optional[float]105        elif isinstance(timeout, tuple):106            # Passed as a tuple.107            self.connect = timeout[0]108            self.read = timeout[1]109            self.write = None if len(timeout) < 3 else timeout[2]110            self.pool = None if len(timeout) < 4 else timeout[3]111        elif not (112            isinstance(connect, UnsetType)113            or isinstance(read, UnsetType)114            or isinstance(write, UnsetType)115            or isinstance(pool, UnsetType)116        ):117            self.connect = connect118            self.read = read119            self.write = write120            self.pool = pool121        else:122            if isinstance(timeout, UnsetType):123                raise ValueError(124                    "httpx.Timeout must either include a default, or set all "125                    "four parameters explicitly."126                )127            self.connect = timeout if isinstance(connect, UnsetType) else connect128            self.read = timeout if isinstance(read, UnsetType) else read129            self.write = timeout if isinstance(write, UnsetType) else write130            self.pool = timeout if isinstance(pool, UnsetType) else pool131 132    def as_dict(self) -> dict[str, float | None]:133        return {134            "connect": self.connect,135            "read": self.read,136            "write": self.write,137            "pool": self.pool,138        }139 140    def __eq__(self, other: typing.Any) -> bool:141        return (142            isinstance(other, self.__class__)143            and self.connect == other.connect144            and self.read == other.read145            and self.write == other.write146            and self.pool == other.pool147        )148 149    def __repr__(self) -> str:150        class_name = self.__class__.__name__151        if len({self.connect, self.read, self.write, self.pool}) == 1:152            return f"{class_name}(timeout={self.connect})"153        return (154            f"{class_name}(connect={self.connect}, "155            f"read={self.read}, write={self.write}, pool={self.pool})"156        )157 158 159class Limits:160    """161    Configuration for limits to various client behaviors.162 163    **Parameters:**164 165    * **max_connections** - The maximum number of concurrent connections that may be166            established.167    * **max_keepalive_connections** - Allow the connection pool to maintain168            keep-alive connections below this point. Should be less than or equal169            to `max_connections`.170    * **keepalive_expiry** - Time limit on idle keep-alive connections in seconds.171    """172 173    def __init__(174        self,175        *,176        max_connections: int | None = None,177        max_keepalive_connections: int | None = None,178        keepalive_expiry: float | None = 5.0,179    ) -> None:180        self.max_connections = max_connections181        self.max_keepalive_connections = max_keepalive_connections182        self.keepalive_expiry = keepalive_expiry183 184    def __eq__(self, other: typing.Any) -> bool:185        return (186            isinstance(other, self.__class__)187            and self.max_connections == other.max_connections188            and self.max_keepalive_connections == other.max_keepalive_connections189            and self.keepalive_expiry == other.keepalive_expiry190        )191 192    def __repr__(self) -> str:193        class_name = self.__class__.__name__194        return (195            f"{class_name}(max_connections={self.max_connections}, "196            f"max_keepalive_connections={self.max_keepalive_connections}, "197            f"keepalive_expiry={self.keepalive_expiry})"198        )199 200 201class Proxy:202    def __init__(203        self,204        url: URL | str,205        *,206        ssl_context: ssl.SSLContext | None = None,207        auth: tuple[str, str] | None = None,208        headers: HeaderTypes | None = None,209    ) -> None:210        url = URL(url)211        headers = Headers(headers)212 213        if url.scheme not in ("http", "https", "socks5", "socks5h"):214            raise ValueError(f"Unknown scheme for proxy URL {url!r}")215 216        if url.username or url.password:217            # Remove any auth credentials from the URL.218            auth = (url.username, url.password)219            url = url.copy_with(username=None, password=None)220 221        self.url = url222        self.auth = auth223        self.headers = headers224        self.ssl_context = ssl_context225 226    @property227    def raw_auth(self) -> tuple[bytes, bytes] | None:228        # The proxy authentication as raw bytes.229        return (230            None231            if self.auth is None232            else (self.auth[0].encode("utf-8"), self.auth[1].encode("utf-8"))233        )234 235    def __repr__(self) -> str:236        # The authentication is represented with the password component masked.237        auth = (self.auth[0], "********") if self.auth else None238 239        # Build a nice concise representation.240        url_str = f"{str(self.url)!r}"241        auth_str = f", auth={auth!r}" if auth else ""242        headers_str = f", headers={dict(self.headers)!r}" if self.headers else ""243        return f"Proxy({url_str}{auth_str}{headers_str})"244 245 246DEFAULT_TIMEOUT_CONFIG = Timeout(timeout=5.0)247DEFAULT_LIMITS = Limits(max_connections=100, max_keepalive_connections=20)248DEFAULT_MAX_REDIRECTS = 20249