"""Support for plugins that provide access to resources suchas schemas."""importpkgutilfromcollections.abcimportMappingfromasdf_standardimportDirectoryResourceMappingas_DirectoryResourceMappingfrom.utilimportget_class_name__all__=["ResourceMappingProxy","DirectoryResourceMapping","ResourceManager","JsonschemaResourceMapping",]
[docs]classDirectoryResourceMapping(_DirectoryResourceMapping):""" A resource mapping that reads resource content from a directory or directory tree. See :class:`~asdf_standard.resource.DirectoryResourceMapping` for details. """
[docs]classResourceMappingProxy(Mapping):""" Wrapper around a resource mapping that carries additional information on the package that provided the mapping. """
def__init__(self,delegate,package_name=None,package_version=None):ifnotisinstance(delegate,Mapping):msg="Resource mapping must implement the Mapping interface"raiseTypeError(msg)self._delegate=delegateself._package_name=package_nameself._package_version=package_versionself._class_name=get_class_name(delegate)def__getitem__(self,uri):returnself._delegate.__getitem__(uri)def__len__(self):returnself._delegate.__len__()def__iter__(self):returnself._delegate.__iter__()@propertydefdelegate(self):""" Get the wrapped mapping instance. Returns ------- collections.abc.Mapping """returnself._delegate@propertydefpackage_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. """returnself._package_name@propertydefpackage_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. """returnself._package_version@propertydefclass_name(self):""" " Get the fully qualified class name of the mapping. Returns ------- str """returnself._class_namedef__eq__(self,other):ifisinstance(other,ResourceMappingProxy):returnother.delegateisself.delegatereturnFalsedef__hash__(self):returnhash(id(self.delegate))def__repr__(self):ifself.package_nameisnotNone:package_description=f"{self.package_name}=={self.package_version}"else:package_description="(none)"returnf"<ResourceMappingProxy class: {self.class_name} package: {package_description} len: {len(self)}>"
[docs]classResourceManager(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_mappingsself._mappings_by_uri={}formappinginresource_mappings:foruriinmapping:ifurinotinself._mappings_by_uri:self._mappings_by_uri[uri]=mappingdef__getitem__(self,uri):ifurinotinself._mappings_by_uri:msg=f"Resource unavailable for URI: {uri}"raiseKeyError(msg)content=self._mappings_by_uri[uri][uri]ifisinstance(content,str):content=content.encode("utf-8")returncontentdef__len__(self):returnlen(self._mappings_by_uri)def__iter__(self):yield fromself._mappings_by_uridef__contains__(self,uri):# Implement __contains__ only for efficiency.returnuriinself._mappings_by_uridef__repr__(self):returnf"<ResourceManager len: {self.__len__()}>"
[docs]classJsonschemaResourceMapping(Mapping):""" Resource mapping that fetches metaschemas from the jsonschema package. """def__getitem__(self,uri):filename=_JSONSCHEMA_URI_TO_FILENAME[uri]returnpkgutil.get_data("asdf._jsonschema",f"schemas/{filename}")def__len__(self):returnlen(_JSONSCHEMA_URI_TO_FILENAME)def__iter__(self):yield from_JSONSCHEMA_URI_TO_FILENAMEdef__repr__(self):return"JsonschemaResourceMapping()"