__init__.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457
  1. """
  2. A Path-like interface for zipfiles.
  3. This codebase is shared between zipfile.Path in the stdlib
  4. and zipp in PyPI. See
  5. https://github.com/python/importlib_metadata/wiki/Development-Methodology
  6. for more detail.
  7. """
  8. import functools
  9. import io
  10. import itertools
  11. import pathlib
  12. import posixpath
  13. import re
  14. import stat
  15. import sys
  16. import zipfile
  17. from ._functools import none_as, save_method_args
  18. from .compat.py310 import text_encoding
  19. from .glob import Translator
  20. __all__ = ['Path']
  21. def _parents(path):
  22. """
  23. Given a path with elements separated by
  24. posixpath.sep, generate all parents of that path.
  25. >>> list(_parents('b/d'))
  26. ['b']
  27. >>> list(_parents('/b/d/'))
  28. ['/b']
  29. >>> list(_parents('b/d/f/'))
  30. ['b/d', 'b']
  31. >>> list(_parents('b'))
  32. []
  33. >>> list(_parents(''))
  34. []
  35. """
  36. return itertools.islice(_ancestry(path), 1, None)
  37. def _ancestry(path):
  38. """
  39. Given a path with elements separated by
  40. posixpath.sep, generate all elements of that path.
  41. >>> list(_ancestry('b/d'))
  42. ['b/d', 'b']
  43. >>> list(_ancestry('/b/d/'))
  44. ['/b/d', '/b']
  45. >>> list(_ancestry('b/d/f/'))
  46. ['b/d/f', 'b/d', 'b']
  47. >>> list(_ancestry('b'))
  48. ['b']
  49. >>> list(_ancestry(''))
  50. []
  51. Multiple separators are treated like a single.
  52. >>> list(_ancestry('//b//d///f//'))
  53. ['//b//d///f', '//b//d', '//b']
  54. """
  55. path = path.rstrip(posixpath.sep)
  56. while path.rstrip(posixpath.sep):
  57. yield path
  58. path, tail = posixpath.split(path)
  59. _dedupe = dict.fromkeys
  60. """Deduplicate an iterable in original order"""
  61. def _difference(minuend, subtrahend):
  62. """
  63. Return items in minuend not in subtrahend, retaining order
  64. with O(1) lookup.
  65. """
  66. return itertools.filterfalse(set(subtrahend).__contains__, minuend)
  67. class InitializedState:
  68. """
  69. Mix-in to save the initialization state for pickling.
  70. """
  71. @save_method_args
  72. def __init__(self, *args, **kwargs):
  73. super().__init__(*args, **kwargs)
  74. def __getstate__(self):
  75. return self._saved___init__.args, self._saved___init__.kwargs
  76. def __setstate__(self, state):
  77. args, kwargs = state
  78. super().__init__(*args, **kwargs)
  79. class CompleteDirs(InitializedState, zipfile.ZipFile):
  80. """
  81. A ZipFile subclass that ensures that implied directories
  82. are always included in the namelist.
  83. >>> list(CompleteDirs._implied_dirs(['foo/bar.txt', 'foo/bar/baz.txt']))
  84. ['foo/', 'foo/bar/']
  85. >>> list(CompleteDirs._implied_dirs(['foo/bar.txt', 'foo/bar/baz.txt', 'foo/bar/']))
  86. ['foo/']
  87. """
  88. @staticmethod
  89. def _implied_dirs(names):
  90. parents = itertools.chain.from_iterable(map(_parents, names))
  91. as_dirs = (p + posixpath.sep for p in parents)
  92. return _dedupe(_difference(as_dirs, names))
  93. def namelist(self):
  94. names = super().namelist()
  95. return names + list(self._implied_dirs(names))
  96. def _name_set(self):
  97. return set(self.namelist())
  98. def resolve_dir(self, name):
  99. """
  100. If the name represents a directory, return that name
  101. as a directory (with the trailing slash).
  102. """
  103. names = self._name_set()
  104. dirname = name + '/'
  105. dir_match = name not in names and dirname in names
  106. return dirname if dir_match else name
  107. def getinfo(self, name):
  108. """
  109. Supplement getinfo for implied dirs.
  110. """
  111. try:
  112. return super().getinfo(name)
  113. except KeyError:
  114. if not name.endswith('/') or name not in self._name_set():
  115. raise
  116. return zipfile.ZipInfo(filename=name)
  117. @classmethod
  118. def make(cls, source):
  119. """
  120. Given a source (filename or zipfile), return an
  121. appropriate CompleteDirs subclass.
  122. """
  123. if isinstance(source, CompleteDirs):
  124. return source
  125. if not isinstance(source, zipfile.ZipFile):
  126. return cls(source)
  127. # Only allow for FastLookup when supplied zipfile is read-only
  128. if 'r' not in source.mode:
  129. cls = CompleteDirs
  130. source.__class__ = cls
  131. return source
  132. @classmethod
  133. def inject(cls, zf: zipfile.ZipFile) -> zipfile.ZipFile:
  134. """
  135. Given a writable zip file zf, inject directory entries for
  136. any directories implied by the presence of children.
  137. """
  138. for name in cls._implied_dirs(zf.namelist()):
  139. zf.writestr(name, b"")
  140. return zf
  141. class FastLookup(CompleteDirs):
  142. """
  143. ZipFile subclass to ensure implicit
  144. dirs exist and are resolved rapidly.
  145. """
  146. def namelist(self):
  147. return self._namelist
  148. @functools.cached_property
  149. def _namelist(self):
  150. return super().namelist()
  151. def _name_set(self):
  152. return self._name_set_prop
  153. @functools.cached_property
  154. def _name_set_prop(self):
  155. return super()._name_set()
  156. def _extract_text_encoding(encoding=None, *args, **kwargs):
  157. # compute stack level so that the caller of the caller sees any warning.
  158. is_pypy = sys.implementation.name == 'pypy'
  159. # PyPy no longer special cased after 7.3.19 (or maybe 7.3.18)
  160. # See jaraco/zipp#143
  161. is_old_pypi = is_pypy and sys.pypy_version_info < (7, 3, 19)
  162. stack_level = 3 + is_old_pypi
  163. return text_encoding(encoding, stack_level), args, kwargs
  164. class Path:
  165. """
  166. A :class:`importlib.resources.abc.Traversable` interface for zip files.
  167. Implements many of the features users enjoy from
  168. :class:`pathlib.Path`.
  169. Consider a zip file with this structure::
  170. .
  171. ├── a.txt
  172. └── b
  173. ├── c.txt
  174. └── d
  175. └── e.txt
  176. >>> data = io.BytesIO()
  177. >>> zf = zipfile.ZipFile(data, 'w')
  178. >>> zf.writestr('a.txt', 'content of a')
  179. >>> zf.writestr('b/c.txt', 'content of c')
  180. >>> zf.writestr('b/d/e.txt', 'content of e')
  181. >>> zf.filename = 'mem/abcde.zip'
  182. Path accepts the zipfile object itself or a filename
  183. >>> path = Path(zf)
  184. From there, several path operations are available.
  185. Directory iteration (including the zip file itself):
  186. >>> a, b = path.iterdir()
  187. >>> a
  188. Path('mem/abcde.zip', 'a.txt')
  189. >>> b
  190. Path('mem/abcde.zip', 'b/')
  191. name property:
  192. >>> b.name
  193. 'b'
  194. join with divide operator:
  195. >>> c = b / 'c.txt'
  196. >>> c
  197. Path('mem/abcde.zip', 'b/c.txt')
  198. >>> c.name
  199. 'c.txt'
  200. Read text:
  201. >>> c.read_text(encoding='utf-8')
  202. 'content of c'
  203. existence:
  204. >>> c.exists()
  205. True
  206. >>> (b / 'missing.txt').exists()
  207. False
  208. Coercion to string:
  209. >>> import os
  210. >>> str(c).replace(os.sep, posixpath.sep)
  211. 'mem/abcde.zip/b/c.txt'
  212. At the root, ``name``, ``filename``, and ``parent``
  213. resolve to the zipfile.
  214. >>> str(path)
  215. 'mem/abcde.zip'
  216. >>> path.name
  217. 'abcde.zip'
  218. >>> path.filename == pathlib.Path('mem/abcde.zip')
  219. True
  220. >>> str(path.parent)
  221. 'mem'
  222. If the zipfile has no filename, such attributes are not
  223. valid and accessing them will raise an Exception.
  224. >>> zf.filename = None
  225. >>> path.name
  226. Traceback (most recent call last):
  227. ...
  228. TypeError: ...
  229. >>> path.filename
  230. Traceback (most recent call last):
  231. ...
  232. TypeError: ...
  233. >>> path.parent
  234. Traceback (most recent call last):
  235. ...
  236. TypeError: ...
  237. # workaround python/cpython#106763
  238. >>> pass
  239. """
  240. __repr = "{self.__class__.__name__}({self.root.filename!r}, {self.at!r})"
  241. def __init__(self, root, at=""):
  242. """
  243. Construct a Path from a ZipFile or filename.
  244. Note: When the source is an existing ZipFile object,
  245. its type (__class__) will be mutated to a
  246. specialized type. If the caller wishes to retain the
  247. original type, the caller should either create a
  248. separate ZipFile object or pass a filename.
  249. """
  250. self.root = FastLookup.make(root)
  251. self.at = at
  252. def __eq__(self, other):
  253. """
  254. >>> Path(zipfile.ZipFile(io.BytesIO(), 'w')) == 'foo'
  255. False
  256. """
  257. if self.__class__ is not other.__class__:
  258. return NotImplemented
  259. return (self.root, self.at) == (other.root, other.at)
  260. def __hash__(self):
  261. return hash((self.root, self.at))
  262. def open(self, mode='r', *args, pwd=None, **kwargs):
  263. """
  264. Open this entry as text or binary following the semantics
  265. of ``pathlib.Path.open()`` by passing arguments through
  266. to io.TextIOWrapper().
  267. """
  268. if self.is_dir():
  269. raise IsADirectoryError(self)
  270. zip_mode = mode[0]
  271. if zip_mode == 'r' and not self.exists():
  272. raise FileNotFoundError(self)
  273. stream = self.root.open(self.at, zip_mode, pwd=pwd)
  274. if 'b' in mode:
  275. if args or kwargs:
  276. raise ValueError("encoding args invalid for binary operation")
  277. return stream
  278. # Text mode:
  279. encoding, args, kwargs = _extract_text_encoding(*args, **kwargs)
  280. return io.TextIOWrapper(stream, encoding, *args, **kwargs)
  281. def _base(self):
  282. return pathlib.PurePosixPath(self.at) if self.at else self.filename
  283. @property
  284. def name(self):
  285. return self._base().name
  286. @property
  287. def suffix(self):
  288. return self._base().suffix
  289. @property
  290. def suffixes(self):
  291. return self._base().suffixes
  292. @property
  293. def stem(self):
  294. return self._base().stem
  295. @property
  296. def filename(self):
  297. return pathlib.Path(self.root.filename).joinpath(self.at)
  298. def read_text(self, *args, **kwargs):
  299. encoding, args, kwargs = _extract_text_encoding(*args, **kwargs)
  300. with self.open('r', encoding, *args, **kwargs) as strm:
  301. return strm.read()
  302. def read_bytes(self):
  303. with self.open('rb') as strm:
  304. return strm.read()
  305. def _is_child(self, path):
  306. return posixpath.dirname(path.at.rstrip("/")) == self.at.rstrip("/")
  307. def _next(self, at):
  308. return self.__class__(self.root, at)
  309. def is_dir(self):
  310. return not self.at or self.at.endswith("/")
  311. def is_file(self):
  312. return self.exists() and not self.is_dir()
  313. def exists(self):
  314. return self.at in self.root._name_set()
  315. def iterdir(self):
  316. if not self.is_dir():
  317. raise ValueError("Can't listdir a file")
  318. subs = map(self._next, self.root.namelist())
  319. return filter(self._is_child, subs)
  320. def match(self, path_pattern):
  321. return pathlib.PurePosixPath(self.at).match(path_pattern)
  322. def is_symlink(self):
  323. """
  324. Return whether this path is a symlink.
  325. """
  326. info = self.root.getinfo(self.at)
  327. mode = info.external_attr >> 16
  328. return stat.S_ISLNK(mode)
  329. def glob(self, pattern):
  330. if not pattern:
  331. raise ValueError(f"Unacceptable pattern: {pattern!r}")
  332. prefix = re.escape(self.at)
  333. tr = Translator(seps='/')
  334. matches = re.compile(prefix + tr.translate(pattern)).fullmatch
  335. return map(self._next, filter(matches, self.root.namelist()))
  336. def rglob(self, pattern):
  337. return self.glob(f'**/{pattern}')
  338. def relative_to(self, other, *extra):
  339. return posixpath.relpath(str(self), str(other.joinpath(*extra)))
  340. def __str__(self):
  341. root = none_as(self.root.filename, ':zipfile:')
  342. return posixpath.join(root, self.at) if self.at else root
  343. def __repr__(self):
  344. return self.__repr.format(self=self)
  345. def joinpath(self, *other):
  346. next = posixpath.join(self.at, *other)
  347. return self._next(self.root.resolve_dir(next))
  348. __truediv__ = joinpath
  349. @property
  350. def parent(self):
  351. if not self.at:
  352. return self.filename.parent
  353. parent_at = posixpath.dirname(self.at.rstrip('/'))
  354. if parent_at:
  355. parent_at += '/'
  356. return self._next(parent_at)