Team Ai
Datasetpublic

codekingpro/portable-devtools

sourceHugging Faceupdated 5mo agoView on Hugging Face
1likes14kdownloads
utils.py756 linesDownload Raw Back to jinja2
1import enum2import json3import os4import re5import typing as t6from collections import abc7from collections import deque8from random import choice9from random import randrange10from threading import Lock11from types import CodeType12from urllib.parse import quote_from_bytes13 14import markupsafe15 16if t.TYPE_CHECKING:17    import typing_extensions as te18 19F = t.TypeVar("F", bound=t.Callable[..., t.Any])20 21# special singleton representing missing values for the runtime22missing: t.Any = type("MissingType", (), {"__repr__": lambda x: "missing"})()23 24internal_code: t.MutableSet[CodeType] = set()25 26concat = "".join27 28 29def pass_context(f: F) -> F:30    """Pass the :class:`~jinja2.runtime.Context` as the first argument31    to the decorated function when called while rendering a template.32 33    Can be used on functions, filters, and tests.34 35    If only ``Context.eval_context`` is needed, use36    :func:`pass_eval_context`. If only ``Context.environment`` is37    needed, use :func:`pass_environment`.38 39    .. versionadded:: 3.0.040        Replaces ``contextfunction`` and ``contextfilter``.41    """42    f.jinja_pass_arg = _PassArg.context  # type: ignore43    return f44 45 46def pass_eval_context(f: F) -> F:47    """Pass the :class:`~jinja2.nodes.EvalContext` as the first argument48    to the decorated function when called while rendering a template.49    See :ref:`eval-context`.50 51    Can be used on functions, filters, and tests.52 53    If only ``EvalContext.environment`` is needed, use54    :func:`pass_environment`.55 56    .. versionadded:: 3.0.057        Replaces ``evalcontextfunction`` and ``evalcontextfilter``.58    """59    f.jinja_pass_arg = _PassArg.eval_context  # type: ignore60    return f61 62 63def pass_environment(f: F) -> F:64    """Pass the :class:`~jinja2.Environment` as the first argument to65    the decorated function when called while rendering a template.66 67    Can be used on functions, filters, and tests.68 69    .. versionadded:: 3.0.070        Replaces ``environmentfunction`` and ``environmentfilter``.71    """72    f.jinja_pass_arg = _PassArg.environment  # type: ignore73    return f74 75 76class _PassArg(enum.Enum):77    context = enum.auto()78    eval_context = enum.auto()79    environment = enum.auto()80 81    @classmethod82    def from_obj(cls, obj: F) -> t.Optional["_PassArg"]:83        if hasattr(obj, "jinja_pass_arg"):84            return obj.jinja_pass_arg  # type: ignore85 86        return None87 88 89def internalcode(f: F) -> F:90    """Marks the function as internally used"""91    internal_code.add(f.__code__)92    return f93 94 95def is_undefined(obj: t.Any) -> bool:96    """Check if the object passed is undefined.  This does nothing more than97    performing an instance check against :class:`Undefined` but looks nicer.98    This can be used for custom filters or tests that want to react to99    undefined variables.  For example a custom default filter can look like100    this::101 102        def default(var, default=''):103            if is_undefined(var):104                return default105            return var106    """107    from .runtime import Undefined108 109    return isinstance(obj, Undefined)110 111 112def consume(iterable: t.Iterable[t.Any]) -> None:113    """Consumes an iterable without doing anything with it."""114    for _ in iterable:115        pass116 117 118def clear_caches() -> None:119    """Jinja keeps internal caches for environments and lexers.  These are120    used so that Jinja doesn't have to recreate environments and lexers all121    the time.  Normally you don't have to care about that but if you are122    measuring memory consumption you may want to clean the caches.123    """124    from .environment import get_spontaneous_environment125    from .lexer import _lexer_cache126 127    get_spontaneous_environment.cache_clear()128    _lexer_cache.clear()129 130 131def import_string(import_name: str, silent: bool = False) -> t.Any:132    """Imports an object based on a string.  This is useful if you want to133    use import paths as endpoints or something similar.  An import path can134    be specified either in dotted notation (``xml.sax.saxutils.escape``)135    or with a colon as object delimiter (``xml.sax.saxutils:escape``).136 137    If the `silent` is True the return value will be `None` if the import138    fails.139 140    :return: imported object141    """142    try:143        if ":" in import_name:144            module, obj = import_name.split(":", 1)145        elif "." in import_name:146            module, _, obj = import_name.rpartition(".")147        else:148            return __import__(import_name)149        return getattr(__import__(module, None, None, [obj]), obj)150    except (ImportError, AttributeError):151        if not silent:152            raise153 154 155def open_if_exists(filename: str, mode: str = "rb") -> t.Optional[t.IO[t.Any]]:156    """Returns a file descriptor for the filename if that file exists,157    otherwise ``None``.158    """159    if not os.path.isfile(filename):160        return None161 162    return open(filename, mode)163 164 165def object_type_repr(obj: t.Any) -> str:166    """Returns the name of the object's type.  For some recognized167    singletons the name of the object is returned instead. (For168    example for `None` and `Ellipsis`).169    """170    if obj is None:171        return "None"172    elif obj is Ellipsis:173        return "Ellipsis"174 175    cls = type(obj)176 177    if cls.__module__ == "builtins":178        return f"{cls.__name__} object"179 180    return f"{cls.__module__}.{cls.__name__} object"181 182 183def pformat(obj: t.Any) -> str:184    """Format an object using :func:`pprint.pformat`."""185    from pprint import pformat186 187    return pformat(obj)188 189 190_http_re = re.compile(191    r"""192    ^193    (194        (https?://|www\.)  # scheme or www195        (([\w%-]+\.)+)?  # subdomain196        (197            [a-z]{2,63}  # basic tld198        |199            xn--[\w%]{2,59}  # idna tld200        )201    |202        ([\w%-]{2,63}\.)+  # basic domain203        (com|net|int|edu|gov|org|info|mil)  # basic tld204    |205        (https?://)  # scheme206        (207            (([\d]{1,3})(\.[\d]{1,3}){3})  # IPv4208        |209            (\[([\da-f]{0,4}:){2}([\da-f]{0,4}:?){1,6}])  # IPv6210        )211    )212    (?::[\d]{1,5})?  # port213    (?:[/?#]\S*)?  # path, query, and fragment214    $215    """,216    re.IGNORECASE | re.VERBOSE,217)218_email_re = re.compile(r"^\S+@\w[\w.-]*\.\w+$")219 220 221def urlize(222    text: str,223    trim_url_limit: t.Optional[int] = None,224    rel: t.Optional[str] = None,225    target: t.Optional[str] = None,226    extra_schemes: t.Optional[t.Iterable[str]] = None,227) -> str:228    """Convert URLs in text into clickable links.229 230    This may not recognize links in some situations. Usually, a more231    comprehensive formatter, such as a Markdown library, is a better232    choice.233 234    Works on ``http://``, ``https://``, ``www.``, ``mailto:``, and email235    addresses. Links with trailing punctuation (periods, commas, closing236    parentheses) and leading punctuation (opening parentheses) are237    recognized excluding the punctuation. Email addresses that include238    header fields are not recognized (for example,239    ``mailto:address@example.com?cc=copy@example.com``).240 241    :param text: Original text containing URLs to link.242    :param trim_url_limit: Shorten displayed URL values to this length.243    :param target: Add the ``target`` attribute to links.244    :param rel: Add the ``rel`` attribute to links.245    :param extra_schemes: Recognize URLs that start with these schemes246        in addition to the default behavior.247 248    .. versionchanged:: 3.0249        The ``extra_schemes`` parameter was added.250 251    .. versionchanged:: 3.0252        Generate ``https://`` links for URLs without a scheme.253 254    .. versionchanged:: 3.0255        The parsing rules were updated. Recognize email addresses with256        or without the ``mailto:`` scheme. Validate IP addresses. Ignore257        parentheses and brackets in more cases.258    """259    if trim_url_limit is not None:260 261        def trim_url(x: str) -> str:262            if len(x) > trim_url_limit:263                return f"{x[:trim_url_limit]}..."264 265            return x266 267    else:268 269        def trim_url(x: str) -> str:270            return x271 272    words = re.split(r"(\s+)", str(markupsafe.escape(text)))273    rel_attr = f' rel="{markupsafe.escape(rel)}"' if rel else ""274    target_attr = f' target="{markupsafe.escape(target)}"' if target else ""275 276    for i, word in enumerate(words):277        head, middle, tail = "", word, ""278        match = re.match(r"^([(<]|&lt;)+", middle)279 280        if match:281            head = match.group()282            middle = middle[match.end() :]283 284        # Unlike lead, which is anchored to the start of the string,285        # need to check that the string ends with any of the characters286        # before trying to match all of them, to avoid backtracking.287        if middle.endswith((")", ">", ".", ",", "\n", "&gt;")):288            match = re.search(r"([)>.,\n]|&gt;)+$", middle)289 290            if match:291                tail = match.group()292                middle = middle[: match.start()]293 294        # Prefer balancing parentheses in URLs instead of ignoring a295        # trailing character.296        for start_char, end_char in ("(", ")"), ("<", ">"), ("&lt;", "&gt;"):297            start_count = middle.count(start_char)298 299            if start_count <= middle.count(end_char):300                # Balanced, or lighter on the left301                continue302 303            # Move as many as possible from the tail to balance304            for _ in range(min(start_count, tail.count(end_char))):305                end_index = tail.index(end_char) + len(end_char)306                # Move anything in the tail before the end char too307                middle += tail[:end_index]308                tail = tail[end_index:]309 310        if _http_re.match(middle):311            if middle.startswith("https://") or middle.startswith("http://"):312                middle = (313                    f'<a href="{middle}"{rel_attr}{target_attr}>{trim_url(middle)}</a>'314                )315            else:316                middle = (317                    f'<a href="https://{middle}"{rel_attr}{target_attr}>'318                    f"{trim_url(middle)}</a>"319                )320 321        elif middle.startswith("mailto:") and _email_re.match(middle[7:]):322            middle = f'<a href="{middle}">{middle[7:]}</a>'323 324        elif (325            "@" in middle326            and not middle.startswith("www.")327            and ":" not in middle328            and _email_re.match(middle)329        ):330            middle = f'<a href="mailto:{middle}">{middle}</a>'331 332        elif extra_schemes is not None:333            for scheme in extra_schemes:334                if middle != scheme and middle.startswith(scheme):335                    middle = f'<a href="{middle}"{rel_attr}{target_attr}>{middle}</a>'336 337        words[i] = f"{head}{middle}{tail}"338 339    return "".join(words)340 341 342def generate_lorem_ipsum(343    n: int = 5, html: bool = True, min: int = 20, max: int = 100344) -> str:345    """Generate some lorem ipsum for the template."""346    from .constants import LOREM_IPSUM_WORDS347 348    words = LOREM_IPSUM_WORDS.split()349    result = []350 351    for _ in range(n):352        next_capitalized = True353        last_comma = last_fullstop = 0354        word = None355        last = None356        p = []357 358        # each paragraph contains out of 20 to 100 words.359        for idx, _ in enumerate(range(randrange(min, max))):360            while True:361                word = choice(words)362                if word != last:363                    last = word364                    break365            if next_capitalized:366                word = word.capitalize()367                next_capitalized = False368            # add commas369            if idx - randrange(3, 8) > last_comma:370                last_comma = idx371                last_fullstop += 2372                word += ","373            # add end of sentences374            if idx - randrange(10, 20) > last_fullstop:375                last_comma = last_fullstop = idx376                word += "."377                next_capitalized = True378            p.append(word)379 380        # ensure that the paragraph ends with a dot.381        p_str = " ".join(p)382 383        if p_str.endswith(","):384            p_str = p_str[:-1] + "."385        elif not p_str.endswith("."):386            p_str += "."387 388        result.append(p_str)389 390    if not html:391        return "\n\n".join(result)392    return markupsafe.Markup(393        "\n".join(f"<p>{markupsafe.escape(x)}</p>" for x in result)394    )395 396 397def url_quote(obj: t.Any, charset: str = "utf-8", for_qs: bool = False) -> str:398    """Quote a string for use in a URL using the given charset.399 400    :param obj: String or bytes to quote. Other types are converted to401        string then encoded to bytes using the given charset.402    :param charset: Encode text to bytes using this charset.403    :param for_qs: Quote "/" and use "+" for spaces.404    """405    if not isinstance(obj, bytes):406        if not isinstance(obj, str):407            obj = str(obj)408 409        obj = obj.encode(charset)410 411    safe = b"" if for_qs else b"/"412    rv = quote_from_bytes(obj, safe)413 414    if for_qs:415        rv = rv.replace("%20", "+")416 417    return rv418 419 420@abc.MutableMapping.register421class LRUCache:422    """A simple LRU Cache implementation."""423 424    # this is fast for small capacities (something below 1000) but doesn't425    # scale.  But as long as it's only used as storage for templates this426    # won't do any harm.427 428    def __init__(self, capacity: int) -> None:429        self.capacity = capacity430        self._mapping: t.Dict[t.Any, t.Any] = {}431        self._queue: "te.Deque[t.Any]" = deque()432        self._postinit()433 434    def _postinit(self) -> None:435        # alias all queue methods for faster lookup436        self._popleft = self._queue.popleft437        self._pop = self._queue.pop438        self._remove = self._queue.remove439        self._wlock = Lock()440        self._append = self._queue.append441 442    def __getstate__(self) -> t.Mapping[str, t.Any]:443        return {444            "capacity": self.capacity,445            "_mapping": self._mapping,446            "_queue": self._queue,447        }448 449    def __setstate__(self, d: t.Mapping[str, t.Any]) -> None:450        self.__dict__.update(d)451        self._postinit()452 453    def __getnewargs__(self) -> t.Tuple[t.Any, ...]:454        return (self.capacity,)455 456    def copy(self) -> "LRUCache":457        """Return a shallow copy of the instance."""458        rv = self.__class__(self.capacity)459        rv._mapping.update(self._mapping)460        rv._queue.extend(self._queue)461        return rv462 463    def get(self, key: t.Any, default: t.Any = None) -> t.Any:464        """Return an item from the cache dict or `default`"""465        try:466            return self[key]467        except KeyError:468            return default469 470    def setdefault(self, key: t.Any, default: t.Any = None) -> t.Any:471        """Set `default` if the key is not in the cache otherwise472        leave unchanged. Return the value of this key.473        """474        try:475            return self[key]476        except KeyError:477            self[key] = default478            return default479 480    def clear(self) -> None:481        """Clear the cache."""482        with self._wlock:483            self._mapping.clear()484            self._queue.clear()485 486    def __contains__(self, key: t.Any) -> bool:487        """Check if a key exists in this cache."""488        return key in self._mapping489 490    def __len__(self) -> int:491        """Return the current size of the cache."""492        return len(self._mapping)493 494    def __repr__(self) -> str:495        return f"<{type(self).__name__} {self._mapping!r}>"496 497    def __getitem__(self, key: t.Any) -> t.Any:498        """Get an item from the cache. Moves the item up so that it has the499        highest priority then.500 501        Raise a `KeyError` if it does not exist.502        """503        with self._wlock:504            rv = self._mapping[key]505 506            if self._queue[-1] != key:507                try:508                    self._remove(key)509                except ValueError:510                    # if something removed the key from the container511                    # when we read, ignore the ValueError that we would512                    # get otherwise.513                    pass514 515                self._append(key)516 517            return rv518 519    def __setitem__(self, key: t.Any, value: t.Any) -> None:520        """Sets the value for an item. Moves the item up so that it521        has the highest priority then.522        """523        with self._wlock:524            if key in self._mapping:525                self._remove(key)526            elif len(self._mapping) == self.capacity:527                del self._mapping[self._popleft()]528 529            self._append(key)530            self._mapping[key] = value531 532    def __delitem__(self, key: t.Any) -> None:533        """Remove an item from the cache dict.534        Raise a `KeyError` if it does not exist.535        """536        with self._wlock:537            del self._mapping[key]538 539            try:540                self._remove(key)541            except ValueError:542                pass543 544    def items(self) -> t.Iterable[t.Tuple[t.Any, t.Any]]:545        """Return a list of items."""546        result = [(key, self._mapping[key]) for key in list(self._queue)]547        result.reverse()548        return result549 550    def values(self) -> t.Iterable[t.Any]:551        """Return a list of all values."""552        return [x[1] for x in self.items()]553 554    def keys(self) -> t.Iterable[t.Any]:555        """Return a list of all keys ordered by most recent usage."""556        return list(self)557 558    def __iter__(self) -> t.Iterator[t.Any]:559        return reversed(tuple(self._queue))560 561    def __reversed__(self) -> t.Iterator[t.Any]:562        """Iterate over the keys in the cache dict, oldest items563        coming first.564        """565        return iter(tuple(self._queue))566 567    __copy__ = copy568 569 570def select_autoescape(571    enabled_extensions: t.Collection[str] = ("html", "htm", "xml"),572    disabled_extensions: t.Collection[str] = (),573    default_for_string: bool = True,574    default: bool = False,575) -> t.Callable[[t.Optional[str]], bool]:576    """Intelligently sets the initial value of autoescaping based on the577    filename of the template.  This is the recommended way to configure578    autoescaping if you do not want to write a custom function yourself.579 580    If you want to enable it for all templates created from strings or581    for all templates with `.html` and `.xml` extensions::582 583        from jinja2 import Environment, select_autoescape584        env = Environment(autoescape=select_autoescape(585            enabled_extensions=('html', 'xml'),586            default_for_string=True,587        ))588 589    Example configuration to turn it on at all times except if the template590    ends with `.txt`::591 592        from jinja2 import Environment, select_autoescape593        env = Environment(autoescape=select_autoescape(594            disabled_extensions=('txt',),595            default_for_string=True,596            default=True,597        ))598 599    The `enabled_extensions` is an iterable of all the extensions that600    autoescaping should be enabled for.  Likewise `disabled_extensions` is601    a list of all templates it should be disabled for.  If a template is602    loaded from a string then the default from `default_for_string` is used.603    If nothing matches then the initial value of autoescaping is set to the604    value of `default`.605 606    For security reasons this function operates case insensitive.607 608    .. versionadded:: 2.9609    """610    enabled_patterns = tuple(f".{x.lstrip('.').lower()}" for x in enabled_extensions)611    disabled_patterns = tuple(f".{x.lstrip('.').lower()}" for x in disabled_extensions)612 613    def autoescape(template_name: t.Optional[str]) -> bool:614        if template_name is None:615            return default_for_string616        template_name = template_name.lower()617        if template_name.endswith(enabled_patterns):618            return True619        if template_name.endswith(disabled_patterns):620            return False621        return default622 623    return autoescape624 625 626def htmlsafe_json_dumps(627    obj: t.Any, dumps: t.Optional[t.Callable[..., str]] = None, **kwargs: t.Any628) -> markupsafe.Markup:629    """Serialize an object to a string of JSON with :func:`json.dumps`,630    then replace HTML-unsafe characters with Unicode escapes and mark631    the result safe with :class:`~markupsafe.Markup`.632 633    This is available in templates as the ``|tojson`` filter.634 635    The following characters are escaped: ``<``, ``>``, ``&``, ``'``.636 637    The returned string is safe to render in HTML documents and638    ``<script>`` tags. The exception is in HTML attributes that are639    double quoted; either use single quotes or the ``|forceescape``640    filter.641 642    :param obj: The object to serialize to JSON.643    :param dumps: The ``dumps`` function to use. Defaults to644        ``env.policies["json.dumps_function"]``, which defaults to645        :func:`json.dumps`.646    :param kwargs: Extra arguments to pass to ``dumps``. Merged onto647        ``env.policies["json.dumps_kwargs"]``.648 649    .. versionchanged:: 3.0650        The ``dumper`` parameter is renamed to ``dumps``.651 652    .. versionadded:: 2.9653    """654    if dumps is None:655        dumps = json.dumps656 657    return markupsafe.Markup(658        dumps(obj, **kwargs)659        .replace("<", "\\u003c")660        .replace(">", "\\u003e")661        .replace("&", "\\u0026")662        .replace("'", "\\u0027")663    )664 665 666class Cycler:667    """Cycle through values by yield them one at a time, then restarting668    once the end is reached. Available as ``cycler`` in templates.669 670    Similar to ``loop.cycle``, but can be used outside loops or across671    multiple loops. For example, render a list of folders and files in a672    list, alternating giving them "odd" and "even" classes.673 674    .. code-block:: html+jinja675 676        {% set row_class = cycler("odd", "even") %}677        <ul class="browser">678        {% for folder in folders %}679          <li class="folder {{ row_class.next() }}">{{ folder }}680        {% endfor %}681        {% for file in files %}682          <li class="file {{ row_class.next() }}">{{ file }}683        {% endfor %}684        </ul>685 686    :param items: Each positional argument will be yielded in the order687        given for each cycle.688 689    .. versionadded:: 2.1690    """691 692    def __init__(self, *items: t.Any) -> None:693        if not items:694            raise RuntimeError("at least one item has to be provided")695        self.items = items696        self.pos = 0697 698    def reset(self) -> None:699        """Resets the current item to the first item."""700        self.pos = 0701 702    @property703    def current(self) -> t.Any:704        """Return the current item. Equivalent to the item that will be705        returned next time :meth:`next` is called.706        """707        return self.items[self.pos]708 709    def next(self) -> t.Any:710        """Return the current item, then advance :attr:`current` to the711        next item.712        """713        rv = self.current714        self.pos = (self.pos + 1) % len(self.items)715        return rv716 717    __next__ = next718 719 720class Joiner:721    """A joining helper for templates."""722 723    def __init__(self, sep: str = ", ") -> None:724        self.sep = sep725        self.used = False726 727    def __call__(self) -> str:728        if not self.used:729            self.used = True730            return ""731        return self.sep732 733 734class Namespace:735    """A namespace object that can hold arbitrary attributes.  It may be736    initialized from a dictionary or with keyword arguments."""737 738    def __init__(*args: t.Any, **kwargs: t.Any) -> None:  # noqa: B902739        self, args = args[0], args[1:]740        self.__attrs = dict(*args, **kwargs)741 742    def __getattribute__(self, name: str) -> t.Any:743        # __class__ is needed for the awaitable check in async mode744        if name in {"_Namespace__attrs", "__class__"}:745            return object.__getattribute__(self, name)746        try:747            return self.__attrs[name]748        except KeyError:749            raise AttributeError(name) from None750 751    def __setitem__(self, name: str, value: t.Any) -> None:752        self.__attrs[name] = value753 754    def __repr__(self) -> str:755        return f"<Namespace {self.__attrs!r}>"756 
codekingpro/portable-devtools · Team Ai