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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
"""
Support for plugins that provide access to resources such
as schemas.
"""
import pkgutil
from collections.abc import Mapping
from asdf_standard import DirectoryResourceMapping as _DirectoryResourceMapping
from .util import get_class_name
__all__ = [
"DirectoryResourceMapping",
"JsonschemaResourceMapping",
"ResourceManager",
"ResourceMappingProxy",
]
class DirectoryResourceMapping(_DirectoryResourceMapping):
"""
A resource mapping that reads resource content from a directory or directory tree.
See :class:`~asdf_standard.resource.DirectoryResourceMapping` for details.
"""
class ResourceMappingProxy(Mapping):
"""
Wrapper around a resource mapping that carries
additional information on the package that provided
the mapping.
"""
@classmethod
def maybe_wrap(cls, delegate):
if isinstance(delegate, ResourceMappingProxy):
return delegate
return ResourceMappingProxy(delegate)
def __init__(self, delegate, package_name=None, package_version=None):
if not isinstance(delegate, Mapping):
msg = "Resource mapping must implement the Mapping interface"
raise TypeError(msg)
self._delegate = delegate
self._package_name = package_name
self._package_version = package_version
self._class_name = get_class_name(delegate)
def __getitem__(self, uri):
return self._delegate.__getitem__(uri)
def __len__(self):
return self._delegate.__len__()
def __iter__(self):
return self._delegate.__iter__()
@property
def delegate(self):
"""
Get the wrapped mapping instance.
Returns
-------
collections.abc.Mapping
"""
return self._delegate
@property
def package_name(self):
"""
Get the name of the Python package that provided this mapping.
Returns
-------
str or None
`None` if the mapping was added at runtime.
"""
return self._package_name
@property
def package_version(self):
"""
Get the version of the Python package that provided the mapping.
Returns
-------
str or None
`None` if the mapping was added at runtime.
"""
return self._package_version
@property
def class_name(self):
""" "
Get the fully qualified class name of the mapping.
Returns
-------
str
"""
return self._class_name
def __eq__(self, other):
if isinstance(other, ResourceMappingProxy):
return other.delegate is self.delegate
return False
def __hash__(self):
return hash(id(self.delegate))
def __repr__(self):
if self.package_name is not None:
package_description = f"{self.package_name}=={self.package_version}"
else:
package_description = "(none)"
return f"<ResourceMappingProxy class: {self.class_name} package: {package_description} len: {len(self)}>"
class ResourceManager(Mapping):
"""
Wraps multiple resource mappings into a single interface
with some friendlier error handling.
Parameters
----------
resource_mappings : iterable of collections.abc.Mapping
Underlying resource mappings. In the case of a duplicate URI,
the first mapping takes precedence.
"""
def __init__(self, resource_mappings):
self._resource_mappings = resource_mappings
self._mappings_by_uri = {}
for mapping in resource_mappings:
for uri in mapping:
if uri not in self._mappings_by_uri:
self._mappings_by_uri[uri] = mapping
def __getitem__(self, uri):
if uri not in self._mappings_by_uri:
msg = f"Resource unavailable for URI: {uri}"
raise KeyError(msg)
content = self._mappings_by_uri[uri][uri]
if isinstance(content, str):
content = content.encode("utf-8")
return content
def __len__(self):
return len(self._mappings_by_uri)
def __iter__(self):
yield from self._mappings_by_uri
def __contains__(self, uri):
# Implement __contains__ only for efficiency.
return uri in self._mappings_by_uri
def __repr__(self):
return f"<ResourceManager len: {self.__len__()}>"
_JSONSCHEMA_URI_TO_FILENAME = {
"http://json-schema.org/draft-04/schema": "draft4.json",
}
class JsonschemaResourceMapping(Mapping):
"""
Resource mapping that fetches metaschemas from
the jsonschema package.
"""
def __getitem__(self, uri):
filename = _JSONSCHEMA_URI_TO_FILENAME[uri]
return pkgutil.get_data("asdf._jsonschema", f"schemas/{filename}")
def __len__(self):
return len(_JSONSCHEMA_URI_TO_FILENAME)
def __iter__(self):
yield from _JSONSCHEMA_URI_TO_FILENAME
def __repr__(self):
return "JsonschemaResourceMapping()"
|
