_adapters.py 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168
  1. from contextlib import suppress
  2. from io import TextIOWrapper
  3. from . import abc
  4. class SpecLoaderAdapter:
  5. """
  6. Adapt a package spec to adapt the underlying loader.
  7. """
  8. def __init__(self, spec, adapter=lambda spec: spec.loader):
  9. self.spec = spec
  10. self.loader = adapter(spec)
  11. def __getattr__(self, name):
  12. return getattr(self.spec, name)
  13. class TraversableResourcesLoader:
  14. """
  15. Adapt a loader to provide TraversableResources.
  16. """
  17. def __init__(self, spec):
  18. self.spec = spec
  19. def get_resource_reader(self, name):
  20. return CompatibilityFiles(self.spec)._native()
  21. def _io_wrapper(file, mode='r', *args, **kwargs):
  22. if mode == 'r':
  23. return TextIOWrapper(file, *args, **kwargs)
  24. elif mode == 'rb':
  25. return file
  26. raise ValueError(f"Invalid mode value '{mode}', only 'r' and 'rb' are supported")
  27. class CompatibilityFiles:
  28. """
  29. Adapter for an existing or non-existent resource reader
  30. to provide a compatibility .files().
  31. """
  32. class SpecPath(abc.Traversable):
  33. """
  34. Path tied to a module spec.
  35. Can be read and exposes the resource reader children.
  36. """
  37. def __init__(self, spec, reader):
  38. self._spec = spec
  39. self._reader = reader
  40. def iterdir(self):
  41. if not self._reader:
  42. return iter(())
  43. return iter(
  44. CompatibilityFiles.ChildPath(self._reader, path)
  45. for path in self._reader.contents()
  46. )
  47. def is_file(self):
  48. return False
  49. is_dir = is_file
  50. def joinpath(self, other):
  51. if not self._reader:
  52. return CompatibilityFiles.OrphanPath(other)
  53. return CompatibilityFiles.ChildPath(self._reader, other)
  54. @property
  55. def name(self):
  56. return self._spec.name
  57. def open(self, mode='r', *args, **kwargs):
  58. return _io_wrapper(self._reader.open_resource(None), mode, *args, **kwargs)
  59. class ChildPath(abc.Traversable):
  60. """
  61. Path tied to a resource reader child.
  62. Can be read but doesn't expose any meaningful children.
  63. """
  64. def __init__(self, reader, name):
  65. self._reader = reader
  66. self._name = name
  67. def iterdir(self):
  68. return iter(())
  69. def is_file(self):
  70. return self._reader.is_resource(self.name)
  71. def is_dir(self):
  72. return not self.is_file()
  73. def joinpath(self, other):
  74. return CompatibilityFiles.OrphanPath(self.name, other)
  75. @property
  76. def name(self):
  77. return self._name
  78. def open(self, mode='r', *args, **kwargs):
  79. return _io_wrapper(
  80. self._reader.open_resource(self.name), mode, *args, **kwargs
  81. )
  82. class OrphanPath(abc.Traversable):
  83. """
  84. Orphan path, not tied to a module spec or resource reader.
  85. Can't be read and doesn't expose any meaningful children.
  86. """
  87. def __init__(self, *path_parts):
  88. if len(path_parts) < 1:
  89. raise ValueError('Need at least one path part to construct a path')
  90. self._path = path_parts
  91. def iterdir(self):
  92. return iter(())
  93. def is_file(self):
  94. return False
  95. is_dir = is_file
  96. def joinpath(self, other):
  97. return CompatibilityFiles.OrphanPath(*self._path, other)
  98. @property
  99. def name(self):
  100. return self._path[-1]
  101. def open(self, mode='r', *args, **kwargs):
  102. raise FileNotFoundError("Can't open orphan path")
  103. def __init__(self, spec):
  104. self.spec = spec
  105. @property
  106. def _reader(self):
  107. with suppress(AttributeError):
  108. return self.spec.loader.get_resource_reader(self.spec.name)
  109. def _native(self):
  110. """
  111. Return the native reader if it supports files().
  112. """
  113. reader = self._reader
  114. return reader if hasattr(reader, 'files') else self
  115. def __getattr__(self, attr):
  116. return getattr(self._reader, attr)
  117. def files(self):
  118. return CompatibilityFiles.SpecPath(self.spec, self._reader)
  119. def wrap_spec(package):
  120. """
  121. Construct a package spec with traversable compatibility
  122. on the spec/loader/reader.
  123. """
  124. return SpecLoaderAdapter(package.__spec__, TraversableResourcesLoader)