codekingpro/portable-devtools

1"""2Helpers related to (dynamic) resource retrieval.3"""4 5from __future__ import annotations6 7from functools import lru_cache8from typing import TYPE_CHECKING9import json10 11try:12    from typing_extensions import TypeVar13except ImportError:  # pragma: no cover14    from typing import TypeVar15 16from referencing import Resource17 18if TYPE_CHECKING:19    from collections.abc import Callable20 21    from referencing.typing import URI, D, Retrieve22 23#: A serialized document (e.g. a JSON string)24_T = TypeVar("_T", default=str)25 26 27def to_cached_resource(28    cache: Callable[[Retrieve[D]], Retrieve[D]] | None = None,29    loads: Callable[[_T], D] = json.loads,30    from_contents: Callable[[D], Resource[D]] = Resource.from_contents,31) -> Callable[[Callable[[URI], _T]], Retrieve[D]]:32    """33    Create a retriever which caches its return values from a simpler callable.34 35    Takes a function which returns things like serialized JSON (strings) and36    returns something suitable for passing to `Registry` as a retrieve37    function.38 39    This decorator both reduces a small bit of boilerplate for a common case40    (deserializing JSON from strings and creating `Resource` objects from the41    result) as well as makes the probable need for caching a bit easier.42    Retrievers which otherwise do expensive operations (like hitting the43    network) might otherwise be called repeatedly.44 45    Examples46    --------47 48    .. testcode::49 50        from referencing import Registry51        from referencing.typing import URI52        import referencing.retrieval53 54 55        @referencing.retrieval.to_cached_resource()56        def retrieve(uri: URI):57            print(f"Retrieved {uri}")58 59            # Normally, go get some expensive JSON from the network, a file ...60            return '''61                {62                    "$schema": "https://json-schema.org/draft/2020-12/schema",63                    "foo": "bar"64                }65            '''66 67        one = Registry(retrieve=retrieve).get_or_retrieve("urn:example:foo")68        print(one.value.contents["foo"])69 70        # Retrieving the same URI again reuses the same value (and thus doesn't71        # print another retrieval message here)72        two = Registry(retrieve=retrieve).get_or_retrieve("urn:example:foo")73        print(two.value.contents["foo"])74 75    .. testoutput::76 77        Retrieved urn:example:foo78        bar79        bar80 81    """82    if cache is None:83        cache = lru_cache(maxsize=None)84 85    def decorator(retrieve: Callable[[URI], _T]):86        @cache87        def cached_retrieve(uri: URI):88            response = retrieve(uri)89            contents = loads(response)90            return from_contents(contents)91 92        return cached_retrieve93 94    return decorator95