Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/zipp/__init__.py: 39%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

181 statements  

1""" 

2A Path-like interface for zipfiles. 

3 

4This codebase is shared between zipfile.Path in the stdlib 

5and zipp in PyPI. See 

6https://github.com/python/importlib_metadata/wiki/Development-Methodology 

7for more detail. 

8""" 

9 

10import io 

11import posixpath 

12import zipfile 

13import itertools 

14import contextlib 

15import pathlib 

16import re 

17import stat 

18import sys 

19 

20from .compat.py310 import text_encoding 

21from .glob import Translator 

22 

23 

24__all__ = ['Path'] 

25 

26 

27def _parents(path): 

28 """ 

29 Given a path with elements separated by 

30 posixpath.sep, generate all parents of that path. 

31 

32 >>> list(_parents('b/d')) 

33 ['b'] 

34 >>> list(_parents('/b/d/')) 

35 ['/b'] 

36 >>> list(_parents('b/d/f/')) 

37 ['b/d', 'b'] 

38 >>> list(_parents('b')) 

39 [] 

40 >>> list(_parents('')) 

41 [] 

42 """ 

43 return itertools.islice(_ancestry(path), 1, None) 

44 

45 

46def _ancestry(path): 

47 """ 

48 Given a path with elements separated by 

49 posixpath.sep, generate all elements of that path. 

50 

51 >>> list(_ancestry('b/d')) 

52 ['b/d', 'b'] 

53 >>> list(_ancestry('/b/d/')) 

54 ['/b/d', '/b'] 

55 >>> list(_ancestry('b/d/f/')) 

56 ['b/d/f', 'b/d', 'b'] 

57 >>> list(_ancestry('b')) 

58 ['b'] 

59 >>> list(_ancestry('')) 

60 [] 

61 

62 Multiple separators are treated like a single. 

63 

64 >>> list(_ancestry('//b//d///f//')) 

65 ['//b//d///f', '//b//d', '//b'] 

66 """ 

67 path = path.rstrip(posixpath.sep) 

68 while path.rstrip(posixpath.sep): 

69 yield path 

70 path, tail = posixpath.split(path) 

71 

72 

73_dedupe = dict.fromkeys 

74"""Deduplicate an iterable in original order""" 

75 

76 

77def _difference(minuend, subtrahend): 

78 """ 

79 Return items in minuend not in subtrahend, retaining order 

80 with O(1) lookup. 

81 """ 

82 return itertools.filterfalse(set(subtrahend).__contains__, minuend) 

83 

84 

85class InitializedState: 

86 """ 

87 Mix-in to save the initialization state for pickling. 

88 """ 

89 

90 def __init__(self, *args, **kwargs): 

91 self.__args = args 

92 self.__kwargs = kwargs 

93 super().__init__(*args, **kwargs) 

94 

95 def __getstate__(self): 

96 return self.__args, self.__kwargs 

97 

98 def __setstate__(self, state): 

99 args, kwargs = state 

100 super().__init__(*args, **kwargs) 

101 

102 

103class CompleteDirs(InitializedState, zipfile.ZipFile): 

104 """ 

105 A ZipFile subclass that ensures that implied directories 

106 are always included in the namelist. 

107 

108 >>> list(CompleteDirs._implied_dirs(['foo/bar.txt', 'foo/bar/baz.txt'])) 

109 ['foo/', 'foo/bar/'] 

110 >>> list(CompleteDirs._implied_dirs(['foo/bar.txt', 'foo/bar/baz.txt', 'foo/bar/'])) 

111 ['foo/'] 

112 """ 

113 

114 @staticmethod 

115 def _implied_dirs(names): 

116 parents = itertools.chain.from_iterable(map(_parents, names)) 

117 as_dirs = (p + posixpath.sep for p in parents) 

118 return _dedupe(_difference(as_dirs, names)) 

119 

120 def namelist(self): 

121 names = super().namelist() 

122 return names + list(self._implied_dirs(names)) 

123 

124 def _name_set(self): 

125 return set(self.namelist()) 

126 

127 def resolve_dir(self, name): 

128 """ 

129 If the name represents a directory, return that name 

130 as a directory (with the trailing slash). 

131 """ 

132 names = self._name_set() 

133 dirname = name + '/' 

134 dir_match = name not in names and dirname in names 

135 return dirname if dir_match else name 

136 

137 def getinfo(self, name): 

138 """ 

139 Supplement getinfo for implied dirs. 

140 """ 

141 try: 

142 return super().getinfo(name) 

143 except KeyError: 

144 if not name.endswith('/') or name not in self._name_set(): 

145 raise 

146 return zipfile.ZipInfo(filename=name) 

147 

148 @classmethod 

149 def make(cls, source): 

150 """ 

151 Given a source (filename or zipfile), return an 

152 appropriate CompleteDirs subclass. 

153 """ 

154 if isinstance(source, CompleteDirs): 

155 return source 

156 

157 if not isinstance(source, zipfile.ZipFile): 

158 return cls(source) 

159 

160 # Only allow for FastLookup when supplied zipfile is read-only 

161 if 'r' not in source.mode: 

162 cls = CompleteDirs 

163 

164 source.__class__ = cls 

165 return source 

166 

167 @classmethod 

168 def inject(cls, zf: zipfile.ZipFile) -> zipfile.ZipFile: 

169 """ 

170 Given a writable zip file zf, inject directory entries for 

171 any directories implied by the presence of children. 

172 """ 

173 for name in cls._implied_dirs(zf.namelist()): 

174 zf.writestr(name, b"") 

175 return zf 

176 

177 

178class FastLookup(CompleteDirs): 

179 """ 

180 ZipFile subclass to ensure implicit 

181 dirs exist and are resolved rapidly. 

182 """ 

183 

184 def namelist(self): 

185 with contextlib.suppress(AttributeError): 

186 return self.__names 

187 self.__names = super().namelist() 

188 return self.__names 

189 

190 def _name_set(self): 

191 with contextlib.suppress(AttributeError): 

192 return self.__lookup 

193 self.__lookup = super()._name_set() 

194 return self.__lookup 

195 

196 

197def _extract_text_encoding(encoding=None, *args, **kwargs): 

198 # compute stack level so that the caller of the caller sees any warning. 

199 is_pypy = sys.implementation.name == 'pypy' 

200 stack_level = 3 + is_pypy 

201 return text_encoding(encoding, stack_level), args, kwargs 

202 

203 

204class Path: 

205 """ 

206 A :class:`importlib.resources.abc.Traversable` interface for zip files. 

207 

208 Implements many of the features users enjoy from 

209 :class:`pathlib.Path`. 

210 

211 Consider a zip file with this structure:: 

212 

213 . 

214 ├── a.txt 

215 └── b 

216 ├── c.txt 

217 └── d 

218 └── e.txt 

219 

220 >>> data = io.BytesIO() 

221 >>> zf = zipfile.ZipFile(data, 'w') 

222 >>> zf.writestr('a.txt', 'content of a') 

223 >>> zf.writestr('b/c.txt', 'content of c') 

224 >>> zf.writestr('b/d/e.txt', 'content of e') 

225 >>> zf.filename = 'mem/abcde.zip' 

226 

227 Path accepts the zipfile object itself or a filename 

228 

229 >>> path = Path(zf) 

230 

231 From there, several path operations are available. 

232 

233 Directory iteration (including the zip file itself): 

234 

235 >>> a, b = path.iterdir() 

236 >>> a 

237 Path('mem/abcde.zip', 'a.txt') 

238 >>> b 

239 Path('mem/abcde.zip', 'b/') 

240 

241 name property: 

242 

243 >>> b.name 

244 'b' 

245 

246 join with divide operator: 

247 

248 >>> c = b / 'c.txt' 

249 >>> c 

250 Path('mem/abcde.zip', 'b/c.txt') 

251 >>> c.name 

252 'c.txt' 

253 

254 Read text: 

255 

256 >>> c.read_text(encoding='utf-8') 

257 'content of c' 

258 

259 existence: 

260 

261 >>> c.exists() 

262 True 

263 >>> (b / 'missing.txt').exists() 

264 False 

265 

266 Coercion to string: 

267 

268 >>> import os 

269 >>> str(c).replace(os.sep, posixpath.sep) 

270 'mem/abcde.zip/b/c.txt' 

271 

272 At the root, ``name``, ``filename``, and ``parent`` 

273 resolve to the zipfile. 

274 

275 >>> str(path) 

276 'mem/abcde.zip/' 

277 >>> path.name 

278 'abcde.zip' 

279 >>> path.filename == pathlib.Path('mem/abcde.zip') 

280 True 

281 >>> str(path.parent) 

282 'mem' 

283 

284 If the zipfile has no filename, such attributes are not 

285 valid and accessing them will raise an Exception. 

286 

287 >>> zf.filename = None 

288 >>> path.name 

289 Traceback (most recent call last): 

290 ... 

291 TypeError: ... 

292 

293 >>> path.filename 

294 Traceback (most recent call last): 

295 ... 

296 TypeError: ... 

297 

298 >>> path.parent 

299 Traceback (most recent call last): 

300 ... 

301 TypeError: ... 

302 

303 # workaround python/cpython#106763 

304 >>> pass 

305 """ 

306 

307 __repr = "{self.__class__.__name__}({self.root.filename!r}, {self.at!r})" 

308 

309 def __init__(self, root, at=""): 

310 """ 

311 Construct a Path from a ZipFile or filename. 

312 

313 Note: When the source is an existing ZipFile object, 

314 its type (__class__) will be mutated to a 

315 specialized type. If the caller wishes to retain the 

316 original type, the caller should either create a 

317 separate ZipFile object or pass a filename. 

318 """ 

319 self.root = FastLookup.make(root) 

320 self.at = at 

321 

322 def __eq__(self, other): 

323 """ 

324 >>> Path(zipfile.ZipFile(io.BytesIO(), 'w')) == 'foo' 

325 False 

326 """ 

327 if self.__class__ is not other.__class__: 

328 return NotImplemented 

329 return (self.root, self.at) == (other.root, other.at) 

330 

331 def __hash__(self): 

332 return hash((self.root, self.at)) 

333 

334 def open(self, mode='r', *args, pwd=None, **kwargs): 

335 """ 

336 Open this entry as text or binary following the semantics 

337 of ``pathlib.Path.open()`` by passing arguments through 

338 to io.TextIOWrapper(). 

339 """ 

340 if self.is_dir(): 

341 raise IsADirectoryError(self) 

342 zip_mode = mode[0] 

343 if not self.exists() and zip_mode == 'r': 

344 raise FileNotFoundError(self) 

345 stream = self.root.open(self.at, zip_mode, pwd=pwd) 

346 if 'b' in mode: 

347 if args or kwargs: 

348 raise ValueError("encoding args invalid for binary operation") 

349 return stream 

350 # Text mode: 

351 encoding, args, kwargs = _extract_text_encoding(*args, **kwargs) 

352 return io.TextIOWrapper(stream, encoding, *args, **kwargs) 

353 

354 def _base(self): 

355 return pathlib.PurePosixPath(self.at or self.root.filename) 

356 

357 @property 

358 def name(self): 

359 return self._base().name 

360 

361 @property 

362 def suffix(self): 

363 return self._base().suffix 

364 

365 @property 

366 def suffixes(self): 

367 return self._base().suffixes 

368 

369 @property 

370 def stem(self): 

371 return self._base().stem 

372 

373 @property 

374 def filename(self): 

375 return pathlib.Path(self.root.filename).joinpath(self.at) 

376 

377 def read_text(self, *args, **kwargs): 

378 encoding, args, kwargs = _extract_text_encoding(*args, **kwargs) 

379 with self.open('r', encoding, *args, **kwargs) as strm: 

380 return strm.read() 

381 

382 def read_bytes(self): 

383 with self.open('rb') as strm: 

384 return strm.read() 

385 

386 def _is_child(self, path): 

387 return posixpath.dirname(path.at.rstrip("/")) == self.at.rstrip("/") 

388 

389 def _next(self, at): 

390 return self.__class__(self.root, at) 

391 

392 def is_dir(self): 

393 return not self.at or self.at.endswith("/") 

394 

395 def is_file(self): 

396 return self.exists() and not self.is_dir() 

397 

398 def exists(self): 

399 return self.at in self.root._name_set() 

400 

401 def iterdir(self): 

402 if not self.is_dir(): 

403 raise ValueError("Can't listdir a file") 

404 subs = map(self._next, self.root.namelist()) 

405 return filter(self._is_child, subs) 

406 

407 def match(self, path_pattern): 

408 return pathlib.PurePosixPath(self.at).match(path_pattern) 

409 

410 def is_symlink(self): 

411 """ 

412 Return whether this path is a symlink. 

413 """ 

414 info = self.root.getinfo(self.at) 

415 mode = info.external_attr >> 16 

416 return stat.S_ISLNK(mode) 

417 

418 def glob(self, pattern): 

419 if not pattern: 

420 raise ValueError(f"Unacceptable pattern: {pattern!r}") 

421 

422 prefix = re.escape(self.at) 

423 tr = Translator(seps='/') 

424 matches = re.compile(prefix + tr.translate(pattern)).fullmatch 

425 return map(self._next, filter(matches, self.root.namelist())) 

426 

427 def rglob(self, pattern): 

428 return self.glob(f'**/{pattern}') 

429 

430 def relative_to(self, other, *extra): 

431 return posixpath.relpath(str(self), str(other.joinpath(*extra))) 

432 

433 def __str__(self): 

434 return posixpath.join(self.root.filename, self.at) 

435 

436 def __repr__(self): 

437 return self.__repr.format(self=self) 

438 

439 def joinpath(self, *other): 

440 next = posixpath.join(self.at, *other) 

441 return self._next(self.root.resolve_dir(next)) 

442 

443 __truediv__ = joinpath 

444 

445 @property 

446 def parent(self): 

447 if not self.at: 

448 return self.filename.parent 

449 parent_at = posixpath.dirname(self.at.rstrip('/')) 

450 if parent_at: 

451 parent_at += '/' 

452 return self._next(parent_at)