One Hat Cyber Team

Your IP: 216.73.216.232
Server IP: 97.74.87.16
Server: Linux 16.87.74.97.host.secureserver.net 5.14.0-503.38.1.el9_5.x86_64 #1 SMP PREEMPT_DYNAMIC Fri Apr 18 08:52:10 EDT 2025 x86_64
Server Software: Apache
PHP Version: 8.2.28
Buat File | Buat Folder
View File Name : retrieval.py

"""
Helpers related to (dynamic) resource retrieval.
"""

from __future__ import annotations

from functools import lru_cache
from typing import TYPE_CHECKING, Callable
import json

try:
    from typing_extensions import TypeVar
except ImportError:  # pragma: no cover
    from typing import TypeVar

from referencing import Resource

if TYPE_CHECKING:
    from referencing.typing import URI, D, Retrieve

#: A serialized document (e.g. a JSON string)
_T = TypeVar("_T", default=str)


def to_cached_resource(
    cache: Callable[[Retrieve[D]], Retrieve[D]] | None = None,
    loads: Callable[[_T], D] = json.loads,
    from_contents: Callable[[D], Resource[D]] = Resource.from_contents,
) -> Callable[[Callable[[URI], _T]], Retrieve[D]]:
    """
    Create a retriever which caches its return values from a simpler callable.

    Takes a function which returns things like serialized JSON (strings) and
    returns something suitable for passing to `Registry` as a retrieve
    function.

    This decorator both reduces a small bit of boilerplate for a common case
    (deserializing JSON from strings and creating `Resource` objects from the
    result) as well as makes the probable need for caching a bit easier.
    Retrievers which otherwise do expensive operations (like hitting the
    network) might otherwise be called repeatedly.

    Examples
    --------

    .. testcode::

        from referencing import Registry
        from referencing.typing import URI
        import referencing.retrieval


        @referencing.retrieval.to_cached_resource()
        def retrieve(uri: URI):
            print(f"Retrieved {uri}")

            # Normally, go get some expensive JSON from the network, a file ...
            return '''
                {
                    "$schema": "https://json-schema.org/draft/2020-12/schema",
                    "foo": "bar"
                }
            '''

        one = Registry(retrieve=retrieve).get_or_retrieve("urn:example:foo")
        print(one.value.contents["foo"])

        # Retrieving the same URI again reuses the same value (and thus doesn't
        # print another retrieval message here)
        two = Registry(retrieve=retrieve).get_or_retrieve("urn:example:foo")
        print(two.value.contents["foo"])

    .. testoutput::

        Retrieved urn:example:foo
        bar
        bar

    """
    if cache is None:
        cache = lru_cache(maxsize=None)

    def decorator(retrieve: Callable[[URI], _T]):
        @cache
        def cached_retrieve(uri: URI):
            response = retrieve(uri)
            contents = loads(response)
            return from_contents(contents)

        return cached_retrieve

    return decorator