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
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
|
import abc
from packaging.specifiers import SpecifierSet
from asdf.util import get_class_name
from ._compressor import Compressor
from ._converter import ConverterProxy
from ._tag import TagDefinition
from ._validator import Validator
class Extension(abc.ABC):
"""
Abstract base class defining an extension to ASDF.
Implementing classes must provide the `extension_uri`.
Other properties are optional.
"""
@classmethod
def __subclasshook__(cls, class_):
if cls is Extension:
return hasattr(class_, "extension_uri")
return NotImplemented # pragma: no cover
@property
@abc.abstractmethod
def extension_uri(self):
"""
Get the URI of the extension to the ASDF Standard implemented
by this class. Note that this may not uniquely identify the
class itself.
Returns
-------
str
"""
@property
def legacy_class_names(self):
"""
Get the set of fully-qualified class names used by older
versions of this extension. This allows a new-style
implementation of an extension to prevent warnings when a
legacy extension is missing.
Returns
-------
iterable of str
"""
return set()
@property
def asdf_standard_requirement(self):
"""
Get the ASDF Standard version requirement for this extension.
Returns
-------
str or None
If str, PEP 440 version specifier.
If None, support all versions.
"""
return
@property
def converters(self):
"""
Get the `asdf.extension.Converter` instances for tags
and Python types supported by this extension.
Returns
-------
iterable of asdf.extension.Converter
"""
return []
@property
def tags(self):
"""
Get the YAML tags supported by this extension.
Returns
-------
iterable of str or asdf.extension.TagDefinition
"""
return []
@property
def compressors(self):
"""
Get the `asdf.extension.Compressor` instances for
compression schemes supported by this extension.
Returns
-------
iterable of asdf.extension.Compressor
"""
return []
@property
def yaml_tag_handles(self):
"""
Get a dictionary of custom yaml TAG handles defined by the extension.
The dictionary key indicates the TAG handles to be placed in the YAML header,
the value defines the string for tag replacement.
See https://yaml.org/spec/1.1/#tag%20shorthand/
Example: ``{"!foo!": "tag:nowhere.org:custom/"}``
Returns
-------
dict
"""
return {}
@property
def validators(self):
"""
Get the `asdf.extension.Validator` instances for additional
schema properties supported by this extension.
Returns
-------
iterable of asdf.extension.Validator
"""
return []
class ExtensionProxy(Extension):
"""
Proxy that wraps an extension, provides default implementations
of optional methods, and carries additional information on the
package that provided the extension.
"""
@classmethod
def maybe_wrap(cls, delegate):
if isinstance(delegate, ExtensionProxy):
return delegate
return ExtensionProxy(delegate)
def __init__(self, delegate, package_name=None, package_version=None):
if not isinstance(delegate, Extension):
msg = "Extension must implement the Extension interface"
raise TypeError(msg)
self._delegate = delegate
self._package_name = package_name
self._package_version = package_version
self._class_name = get_class_name(delegate)
self._legacy = False
# Sort these out up-front so that errors are raised when the extension is loaded
# and not in the middle of the user's session. The extension will fail to load
# and a warning will be emitted, but it won't crash the program.
self._legacy_class_names = set()
for class_name in getattr(self._delegate, "legacy_class_names", []):
if isinstance(class_name, str):
self._legacy_class_names.add(class_name)
else:
msg = "Extension property 'legacy_class_names' must contain str values"
raise TypeError(msg)
if self._legacy:
self._legacy_class_names.add(self._class_name)
value = getattr(self._delegate, "asdf_standard_requirement", None)
if isinstance(value, str):
self._asdf_standard_requirement = SpecifierSet(value)
elif value is None:
self._asdf_standard_requirement = SpecifierSet()
else:
msg = "Extension property 'asdf_standard_requirement' must be str or None"
raise TypeError(msg)
self._tags = []
for tag in getattr(self._delegate, "tags", []):
if isinstance(tag, str):
self._tags.append(TagDefinition(tag))
elif isinstance(tag, TagDefinition):
self._tags.append(tag)
else:
msg = "Extension property 'tags' must contain str or asdf.extension.TagDefinition values"
raise TypeError(msg)
self._yaml_tag_handles = getattr(delegate, "yaml_tag_handles", {})
# Process the converters last, since they expect ExtensionProxy
# properties to already be available.
self._converters = [ConverterProxy(c, self) for c in getattr(self._delegate, "converters", [])]
self._compressors = []
if hasattr(self._delegate, "compressors"):
for compressor in self._delegate.compressors:
if not isinstance(compressor, Compressor):
msg = "Extension property 'compressors' must contain instances of asdf.extension.Compressor"
raise TypeError(msg)
self._compressors.append(compressor)
self._validators = []
if hasattr(self._delegate, "validators"):
for validator in self._delegate.validators:
if not isinstance(validator, Validator):
msg = "Extension property 'validators' must contain instances of asdf.extension.Validator"
raise TypeError(msg)
self._validators.append(validator)
@property
def extension_uri(self):
"""
Get the URI of the extension to the ASDF Standard implemented
by this class. Note that this may not uniquely identify the
class itself.
Returns
-------
str or None
"""
return getattr(self._delegate, "extension_uri", None)
@property
def legacy_class_names(self):
"""
Get the set of fully-qualified class names used by older
versions of this extension. This allows a new-style
implementation of an extension to prevent warnings when a
legacy extension is missing.
Returns
-------
set of str
"""
return self._legacy_class_names
@property
def asdf_standard_requirement(self):
"""
Get the extension's ASDF Standard requirement.
Returns
-------
packaging.specifiers.SpecifierSet
"""
return self._asdf_standard_requirement
@property
def converters(self):
"""
Get the extension's converters.
Returns
-------
list of asdf.extension.Converter
"""
return self._converters
@property
def compressors(self):
"""
Get the extension's compressors.
Returns
-------
list of asdf.extension.Compressor
"""
return self._compressors
@property
def tags(self):
"""
Get the YAML tags supported by this extension.
Returns
-------
list of asdf.extension.TagDefinition
"""
return self._tags
@property
def types(self):
"""
Get the legacy extension's ExtensionType subclasses.
Returns
-------
iterable of asdf.type.ExtensionType
"""
return getattr(self._delegate, "types", [])
@property
def tag_mapping(self):
"""
Get the legacy extension's tag-to-schema-URI mapping.
Returns
-------
iterable of tuple or callable
"""
return getattr(self._delegate, "tag_mapping", [])
@property
def url_mapping(self):
"""
Get the legacy extension's schema-URI-to-URL mapping.
Returns
-------
iterable of tuple or callable
"""
return getattr(self._delegate, "url_mapping", [])
@property
def delegate(self):
"""
Get the wrapped extension instance.
Returns
-------
asdf.extension.Extension
"""
return self._delegate
@property
def package_name(self):
"""
Get the name of the Python package that provided this extension.
Returns
-------
str or None
`None` if the extension was added at runtime.
"""
return self._package_name
@property
def package_version(self):
"""
Get the version of the Python package that provided the extension
Returns
-------
str or None
`None` if the extension was added at runtime.
"""
return self._package_version
@property
def class_name(self):
"""
Get the fully qualified class name of the extension.
Returns
-------
str
"""
return self._class_name
@property
def legacy(self):
"""
False
"""
return self._legacy
@property
def yaml_tag_handles(self):
"""
Get a dictionary of custom yaml TAG handles defined by the extension.
The dictionary key indicates the TAG handles to be placed in the YAML header,
the value defines the string for tag replacement.
See https://yaml.org/spec/1.1/#tag%20shorthand/
Example: ``{"!foo!": "tag:nowhere.org:custom/"}``
Returns
-------
dict
"""
return self._yaml_tag_handles
@property
def validators(self):
"""
Get the `asdf.extension.Validator` instances for additional
schema properties supported by this extension.
Returns
-------
list of asdf.extension.Validator
"""
return self._validators
def __eq__(self, other):
if isinstance(other, ExtensionProxy):
return other.delegate is self.delegate
return False
def __hash__(self):
return hash(id(self.delegate))
def __repr__(self):
package_description = "(none)" if self.package_name is None else f"{self.package_name}=={self.package_version}"
uri_description = "(none)" if self.extension_uri is None else self.extension_uri
return (
f"<ExtensionProxy URI: {uri_description} class: {self.class_name} "
f"package: {package_description} legacy: {self.legacy}>"
)
|
