1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
"""
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
|
