codekingpro/portable-devtools

1import os2import base643import json4import mimetypes5import re6import sys7import time8import tempfile9import sysconfig10 11from collections import OrderedDict12from contextlib import contextmanager13from http.cookiejar import parse_ns_headers14from pathlib import Path15from pprint import pformat16from urllib.parse import urlsplit17from typing import Any, List, Optional, Tuple, Generator, Callable, Iterable, IO, TypeVar18 19import requests.auth20 21RE_COOKIE_SPLIT = re.compile(r', (?=[^ ;]+=)')22Item = Tuple[str, Any]23Items = List[Item]24T = TypeVar("T")25 26 27class JsonDictPreservingDuplicateKeys(OrderedDict):28    """A specialized JSON dict preserving duplicate keys."""29 30    # Python versions prior to 3.8 suffer from an issue with multiple keys with the same name.31    # `json.dumps(obj, indent=N, sort_keys=True)` will output sorted keys when they are unique, and32    # duplicate keys will be outputted as they were defined in the original data.33    # See <https://bugs.python.org/issue23493#msg400929> for the behavior change between Python versions.34    SUPPORTS_SORTING = sys.version_info >= (3, 8)35 36    def __init__(self, items: Items):37        self._items = items38        self._ensure_items_used()39 40    def _ensure_items_used(self) -> None:41        """HACK: Force `json.dumps()` to use `self.items()` instead of an empty dict.42 43        Two JSON encoders are available on CPython: pure-Python (1) and C (2) implementations.44 45        (1) The pure-python implementation will do a simple `if not dict: return '{}'`,46        and we could fake that check by implementing the `__bool__()` method.47        Source:48            - <https://github.com/python/cpython/blob/9d318ad/Lib/json/encoder.py#L334-L336>49 50        (2) On the other hand, the C implementation will do a check on the number of51        items contained inside the dict, using a verification on `dict->ma_used`, which52        is updated only when an item is added/removed from the dict. For that case,53        there is no workaround but to add an item into the dict.54        Sources:55            - <https://github.com/python/cpython/blob/9d318ad/Modules/_json.c#L1581-L1582>56            - <https://github.com/python/cpython/blob/9d318ad/Include/cpython/dictobject.h#L53>57            - <https://github.com/python/cpython/blob/9d318ad/Include/cpython/dictobject.h#L17-L18>58 59        To please both implementations, we simply add one item to the dict.60 61        """62        if self._items:63            self['__hack__'] = '__hack__'64 65    def items(self) -> Items:66        """Return all items, duplicate ones included.67 68        """69        return self._items70 71 72def load_json_preserve_order_and_dupe_keys(s):73    return json.loads(s, object_pairs_hook=JsonDictPreservingDuplicateKeys)74 75 76def repr_dict(d: dict) -> str:77    return pformat(d)78 79 80def humanize_bytes(n, precision=2):81    # Author: Doug Latornell82    # Licence: MIT83    # URL: https://code.activestate.com/recipes/577081/84    """Return a humanized string representation of a number of bytes.85 86    >>> humanize_bytes(1)87    '1 B'88    >>> humanize_bytes(1024, precision=1)89    '1.0 kB'90    >>> humanize_bytes(1024 * 123, precision=1)91    '123.0 kB'92    >>> humanize_bytes(1024 * 12342, precision=1)93    '12.1 MB'94    >>> humanize_bytes(1024 * 12342, precision=2)95    '12.05 MB'96    >>> humanize_bytes(1024 * 1234, precision=2)97    '1.21 MB'98    >>> humanize_bytes(1024 * 1234 * 1111, precision=2)99    '1.31 GB'100    >>> humanize_bytes(1024 * 1234 * 1111, precision=1)101    '1.3 GB'102 103    """104    abbrevs = [105        (1 << 50, 'PB'),106        (1 << 40, 'TB'),107        (1 << 30, 'GB'),108        (1 << 20, 'MB'),109        (1 << 10, 'kB'),110        (1, 'B')111    ]112 113    if n == 1:114        return '1 B'115 116    for factor, suffix in abbrevs:117        if n >= factor:118            break119 120    # noinspection PyUnboundLocalVariable121    return f'{n / factor:.{precision}f} {suffix}'122 123 124class ExplicitNullAuth(requests.auth.AuthBase):125    """Forces requests to ignore the ``.netrc``.126    <https://github.com/psf/requests/issues/2773#issuecomment-174312831>127    """128 129    def __call__(self, r):130        return r131 132 133def get_content_type(filename):134    """135    Return the content type for ``filename`` in format appropriate136    for Content-Type headers, or ``None`` if the file type is unknown137    to ``mimetypes``.138 139    """140    return mimetypes.guess_type(filename, strict=False)[0]141 142 143def split_cookies(cookies):144    """145    When ``requests`` stores cookies in ``response.headers['Set-Cookie']``146    it concatenates all of them through ``, ``.147 148    This function splits cookies apart being careful to not to149    split on ``, `` which may be part of cookie value.150    """151    if not cookies:152        return []153    return RE_COOKIE_SPLIT.split(cookies)154 155 156def get_expired_cookies(157    cookies: str,158    now: float = None159) -> List[dict]:160 161    now = now or time.time()162 163    def is_expired(expires: Optional[float]) -> bool:164        return expires is not None and expires <= now165 166    attr_sets: List[Tuple[str, str]] = parse_ns_headers(167        split_cookies(cookies)168    )169 170    cookies = [171        # The first attr name is the cookie name.172        dict(attrs[1:], name=attrs[0][0])173        for attrs in attr_sets174    ]175 176    _max_age_to_expires(cookies=cookies, now=now)177 178    return [179        {180            'name': cookie['name'],181            'path': cookie.get('path', '/')182        }183        for cookie in cookies184        if is_expired(expires=cookie.get('expires'))185    ]186 187 188def _max_age_to_expires(cookies, now):189    """190    Translate `max-age` into `expires` for Requests to take it into account.191 192    HACK/FIXME: <https://github.com/psf/requests/issues/5743>193 194    """195    for cookie in cookies:196        if 'expires' in cookie:197            continue198        max_age = cookie.get('max-age')199        if max_age and max_age.isdigit():200            cookie['expires'] = now + float(max_age)201 202 203def parse_content_type_header(header):204    """Borrowed from requests."""205    tokens = header.split(';')206    content_type, params = tokens[0].strip(), tokens[1:]207    params_dict = {}208    items_to_strip = "\"' "209    for param in params:210        param = param.strip()211        if param:212            key, value = param, True213            index_of_equals = param.find("=")214            if index_of_equals != -1:215                key = param[:index_of_equals].strip(items_to_strip)216                value = param[index_of_equals + 1:].strip(items_to_strip)217            params_dict[key.lower()] = value218    return content_type, params_dict219 220 221def as_site(path: Path, **extra_vars) -> Path:222    site_packages_path = sysconfig.get_path(223        'purelib',224        vars={'base': str(path), **extra_vars}225    )226    return Path(site_packages_path)227 228 229def get_site_paths(path: Path) -> Iterable[Path]:230    from httpie.compat import (231        MIN_SUPPORTED_PY_VERSION,232        MAX_SUPPORTED_PY_VERSION,233        is_frozen234    )235 236    if is_frozen:237        [major, min_minor] = MIN_SUPPORTED_PY_VERSION238        [major, max_minor] = MAX_SUPPORTED_PY_VERSION239        for minor in range(min_minor, max_minor + 1):240            yield as_site(241                path,242                py_version_short=f'{major}.{minor}'243            )244    else:245        yield as_site(path)246 247 248def split_iterable(iterable: Iterable[T], key: Callable[[T], bool]) -> Tuple[List[T], List[T]]:249    left, right = [], []250    for item in iterable:251        if key(item):252            left.append(item)253        else:254            right.append(item)255    return left, right256 257 258def unwrap_context(exc: Exception) -> Optional[Exception]:259    context = exc.__context__260    if isinstance(context, Exception):261        return unwrap_context(context)262    else:263        return exc264 265 266def url_as_host(url: str) -> str:267    return urlsplit(url).netloc.split('@')[-1]268 269 270class LockFileError(ValueError):271    pass272 273 274@contextmanager275def open_with_lockfile(file: Path, *args, **kwargs) -> Generator[IO[Any], None, None]:276    file_id = base64.b64encode(os.fsencode(file)).decode()277    target_file = Path(tempfile.gettempdir()) / file_id278 279    # Have an atomic-like touch here, so we'll tighten the possibility of280    # a race occurring between multiple processes accessing the same file.281    try:282        target_file.touch(exist_ok=False)283    except FileExistsError as exc:284        raise LockFileError("Can't modify a locked file.") from exc285 286    try:287        with open(file, *args, **kwargs) as stream:288            yield stream289    finally:290        target_file.unlink()291 292 293def is_version_greater(version_1: str, version_2: str) -> bool:294    # In an ideal scenario, we would depend on `packaging` in order295    # to offer PEP 440 compatible parsing. But since it might not be296    # commonly available for outside packages, and since we are only297    # going to parse HTTPie's own version it should be fine to compare298    # this in a SemVer subset fashion.299 300    def split_version(version: str) -> Tuple[int, ...]:301        parts = []302        for part in version.split('.')[:3]:303            try:304                parts.append(int(part))305            except ValueError:306                break307        return tuple(parts)308 309    return split_version(version_1) > split_version(version_2)310