Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/fsspec/asyn.py: 27%

1import asyncio 

2import asyncio.events 

3import functools 

4import inspect 

5import io 

6import numbers 

7import os 

8import re 

9import threading 

10from glob import has_magic 

11from typing import TYPE_CHECKING, Iterable 

12 

13from .callbacks import DEFAULT_CALLBACK 

14from .exceptions import FSTimeoutError 

15from .implementations.local import LocalFileSystem, make_path_posix, trailing_sep 

16from .spec import AbstractBufferedFile, AbstractFileSystem 

17from .utils import glob_translate, is_exception, other_paths 

18 

19private = re.compile("_[^_]") 

20iothread = [None] # dedicated fsspec IO thread 

21loop = [None] # global event loop for any non-async instance 

22_lock = None # global lock placeholder 

23get_running_loop = asyncio.get_running_loop 

24 

25 

26def get_lock(): 

27 """Allocate or return a threading lock. 

28 

29 The lock is allocated on first use to allow setting one lock per forked process. 

30 """ 

31 global _lock 

32 if not _lock: 

33 _lock = threading.Lock() 

34 return _lock 

35 

36 

37def reset_lock(): 

38 """Reset the global lock. 

39 

40 This should be called only on the init of a forked process to reset the lock to 

41 None, enabling the new forked process to get a new lock. 

42 """ 

43 global _lock 

44 

45 iothread[0] = None 

46 loop[0] = None 

47 _lock = None 

48 

49 

50async def _runner(event, coro, result, timeout=None): 

51 timeout = timeout if timeout else None # convert 0 or 0.0 to None 

52 if timeout is not None: 

53 coro = asyncio.wait_for(coro, timeout=timeout) 

54 try: 

55 result[0] = await coro 

56 except Exception as ex: 

57 result[0] = ex 

58 finally: 

59 event.set() 

60 

61 

62def sync(loop, func, *args, timeout=None, **kwargs): 

63 """ 

64 Make loop run coroutine until it returns. Runs in other thread 

65 

66 Examples 

67 -------- 

68 >>> fsspec.asyn.sync(fsspec.asyn.get_loop(), func, *args, 

69 timeout=timeout, **kwargs) 

70 """ 

71 timeout = timeout if timeout else None # convert 0 or 0.0 to None 

72 # NB: if the loop is not running *yet*, it is OK to submit work 

73 # and we will wait for it 

74 if loop is None or loop.is_closed(): 

75 raise RuntimeError("Loop is not running") 

76 try: 

77 loop0 = asyncio.events.get_running_loop() 

78 if loop0 is loop: 

79 raise NotImplementedError("Calling sync() from within a running loop") 

80 except NotImplementedError: 

81 raise 

82 except RuntimeError: 

83 pass 

84 coro = func(*args, **kwargs) 

85 result = [None] 

86 event = threading.Event() 

87 asyncio.run_coroutine_threadsafe(_runner(event, coro, result, timeout), loop) 

88 while True: 

89 # this loops allows thread to get interrupted 

90 if event.wait(1): 

91 break 

92 if timeout is not None: 

93 timeout -= 1 

94 if timeout < 0: 

95 raise FSTimeoutError 

96 

97 return_result = result[0] 

98 if isinstance(return_result, asyncio.TimeoutError): 

99 # suppress asyncio.TimeoutError, raise FSTimeoutError 

100 raise FSTimeoutError from return_result 

101 elif isinstance(return_result, BaseException): 

102 raise return_result 

103 else: 

104 return return_result 

105 

106 

107def sync_wrapper(func, obj=None): 

108 """Given a function, make so can be called in blocking contexts 

109 

110 Leave obj=None if defining within a class. Pass the instance if attaching 

111 as an attribute of the instance. 

112 """ 

113 

114 @functools.wraps(func) 

115 def wrapper(*args, **kwargs): 

116 self = obj or args[0] 

117 return sync(self.loop, func, *args, **kwargs) 

118 

119 return wrapper 

120 

121 

122def get_loop(): 

123 """Create or return the default fsspec IO loop 

124 

125 The loop will be running on a separate thread. 

126 """ 

127 if loop[0] is None: 

128 with get_lock(): 

129 # repeat the check just in case the loop got filled between the 

130 # previous two calls from another thread 

131 if loop[0] is None: 

132 loop[0] = asyncio.new_event_loop() 

133 th = threading.Thread(target=loop[0].run_forever, name="fsspecIO") 

134 th.daemon = True 

135 th.start() 

136 iothread[0] = th 

137 return loop[0] 

138 

139 

140def reset_after_fork(): 

141 global lock 

142 loop[0] = None 

143 iothread[0] = None 

144 lock = None 

145 

146 

147if hasattr(os, "register_at_fork"): 

148 # should be posix; this will do nothing for spawn or forkserver subprocesses 

149 os.register_at_fork(after_in_child=reset_after_fork) 

150 

151 

152if TYPE_CHECKING: 

153 import resource 

154 

155 ResourceError = resource.error 

156else: 

157 try: 

158 import resource 

159 except ImportError: 

160 resource = None 

161 ResourceError = OSError 

162 else: 

163 ResourceError = getattr(resource, "error", OSError) 

164 

165_DEFAULT_BATCH_SIZE = 128 

166_NOFILES_DEFAULT_BATCH_SIZE = 1280 

167 

168 

169def _get_batch_size(nofiles=False): 

170 from fsspec.config import conf 

171 

172 if nofiles: 

173 if "nofiles_gather_batch_size" in conf: 

174 return conf["nofiles_gather_batch_size"] 

175 else: 

176 if "gather_batch_size" in conf: 

177 return conf["gather_batch_size"] 

178 if nofiles: 

179 return _NOFILES_DEFAULT_BATCH_SIZE 

180 if resource is None: 

181 return _DEFAULT_BATCH_SIZE 

182 

183 try: 

184 soft_limit, _ = resource.getrlimit(resource.RLIMIT_NOFILE) 

185 except (ImportError, ValueError, ResourceError): 

186 return _DEFAULT_BATCH_SIZE 

187 

188 if soft_limit == resource.RLIM_INFINITY: 

189 return -1 

190 else: 

191 return soft_limit // 8 

192 

193 

194def running_async() -> bool: 

195 """Being executed by an event loop?""" 

196 try: 

197 asyncio.get_running_loop() 

198 return True 

199 except RuntimeError: 

200 return False 

201 

202 

203async def _run_coros_in_chunks( 

204 coros, 

205 batch_size=None, 

206 callback=DEFAULT_CALLBACK, 

207 timeout=None, 

208 return_exceptions=False, 

209 nofiles=False, 

210): 

211 """Run the given coroutines in chunks. 

212 

213 Parameters 

214 ---------- 

215 coros: list of coroutines to run 

216 batch_size: int or None 

217 Number of coroutines to submit/wait on simultaneously. 

218 If -1, then it will not be any throttling. If 

219 None, it will be inferred from _get_batch_size() 

220 callback: fsspec.callbacks.Callback instance 

221 Gets a relative_update when each coroutine completes 

222 timeout: number or None 

223 If given, each coroutine times out after this time. Note that, since 

224 there are multiple batches, the total run time of this function will in 

225 general be longer 

226 return_exceptions: bool 

227 Same meaning as in asyncio.gather 

228 nofiles: bool 

229 If inferring the batch_size, does this operation involve local files? 

230 If yes, you normally expect smaller batches. 

231 """ 

232 

233 if batch_size is None: 

234 batch_size = _get_batch_size(nofiles=nofiles) 

235 

236 if batch_size == -1: 

237 batch_size = len(coros) 

238 

239 assert batch_size > 0 

240 

241 async def _run_coro(coro, i): 

242 try: 

243 return await asyncio.wait_for(coro, timeout=timeout), i 

244 except Exception as e: 

245 if not return_exceptions: 

246 raise 

247 return e, i 

248 finally: 

249 callback.relative_update(1) 

250 

251 i = 0 

252 n = len(coros) 

253 results = [None] * n 

254 pending = set() 

255 

256 while pending or i < n: 

257 while len(pending) < batch_size and i < n: 

258 pending.add(asyncio.ensure_future(_run_coro(coros[i], i))) 

259 i += 1 

260 

261 if not pending: 

262 break 

263 

264 done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED) 

265 while done: 

266 result, k = await done.pop() 

267 results[k] = result 

268 

269 return results 

270 

271 

272# these methods should be implemented as async by any async-able backend 

273async_methods = [ 

274 "_ls", 

275 "_cat_file", 

276 "_get_file", 

277 "_put_file", 

278 "_rm_file", 

279 "_cp_file", 

280 "_pipe_file", 

281 "_expand_path", 

282 "_info", 

283 "_isfile", 

284 "_isdir", 

285 "_exists", 

286 "_walk", 

287 "_glob", 

288 "_find", 

289 "_du", 

290 "_size", 

291 "_mkdir", 

292 "_makedirs", 

293] 

294 

295 

296class AsyncFileSystem(AbstractFileSystem): 

297 """Async file operations, default implementations 

298 

299 Passes bulk operations to asyncio.gather for concurrent operation. 

300 

301 Implementations that have concurrent batch operations and/or async methods 

302 should inherit from this class instead of AbstractFileSystem. Docstrings are 

303 copied from the un-underscored method in AbstractFileSystem, if not given. 

304 """ 

305 

306 # note that methods do not have docstring here; they will be copied 

307 # for _* methods and inferred for overridden methods. 

308 

309 async_impl = True 

310 mirror_sync_methods = True 

311 disable_throttling = False 

312 

313 def __init__(self, *args, asynchronous=False, loop=None, batch_size=None, **kwargs): 

314 self.asynchronous = asynchronous 

315 self._pid = os.getpid() 

316 if not asynchronous: 

317 self._loop = loop or get_loop() 

318 else: 

319 self._loop = None 

320 self.batch_size = batch_size 

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

322 

323 @property 

324 def loop(self): 

325 if self._pid != os.getpid(): 

326 raise RuntimeError("This class is not fork-safe") 

327 return self._loop 

328 

329 async def _rm_file(self, path, **kwargs): 

330 raise NotImplementedError 

331 

332 async def _rm(self, path, recursive=False, batch_size=None, **kwargs): 

333 # TODO: implement on_error 

334 batch_size = batch_size or self.batch_size 

335 path = await self._expand_path(path, recursive=recursive) 

336 return await _run_coros_in_chunks( 

337 [self._rm_file(p, **kwargs) for p in reversed(path)], 

338 batch_size=batch_size, 

339 nofiles=True, 

340 ) 

341 

342 async def _cp_file(self, path1, path2, **kwargs): 

343 raise NotImplementedError 

344 

345 async def _mv_file(self, path1, path2): 

346 await self._cp_file(path1, path2) 

347 await self._rm_file(path1) 

348 

349 async def _copy( 

350 self, 

351 path1, 

352 path2, 

353 recursive=False, 

354 on_error=None, 

355 maxdepth=None, 

356 batch_size=None, 

357 **kwargs, 

358 ): 

359 if on_error is None and recursive: 

360 on_error = "ignore" 

361 elif on_error is None: 

362 on_error = "raise" 

363 

364 if isinstance(path1, list) and isinstance(path2, list): 

365 # No need to expand paths when both source and destination 

366 # are provided as lists 

367 paths1 = path1 

368 paths2 = path2 

369 else: 

370 source_is_str = isinstance(path1, str) 

371 paths1 = await self._expand_path( 

372 path1, maxdepth=maxdepth, recursive=recursive 

373 ) 

374 if source_is_str and (not recursive or maxdepth is not None): 

375 # Non-recursive glob does not copy directories 

376 paths1 = [ 

377 p for p in paths1 if not (trailing_sep(p) or await self._isdir(p)) 

378 ] 

379 if not paths1: 

380 return 

381 

382 source_is_file = len(paths1) == 1 

383 dest_is_dir = isinstance(path2, str) and ( 

384 trailing_sep(path2) or await self._isdir(path2) 

385 ) 

386 

387 exists = source_is_str and ( 

388 (has_magic(path1) and source_is_file) 

389 or (not has_magic(path1) and dest_is_dir and not trailing_sep(path1)) 

390 ) 

391 paths2 = other_paths( 

392 paths1, 

393 path2, 

394 exists=exists, 

395 flatten=not source_is_str, 

396 ) 

397 

398 batch_size = batch_size or self.batch_size 

399 coros = [self._cp_file(p1, p2, **kwargs) for p1, p2 in zip(paths1, paths2)] 

400 result = await _run_coros_in_chunks( 

401 coros, batch_size=batch_size, return_exceptions=True, nofiles=True 

402 ) 

403 

404 for ex in filter(is_exception, result): 

405 if on_error == "ignore" and isinstance(ex, FileNotFoundError): 

406 continue 

407 raise ex 

408 

409 async def _pipe_file(self, path, value, mode="overwrite", **kwargs): 

410 raise NotImplementedError 

411 

412 async def _pipe(self, path, value=None, batch_size=None, **kwargs): 

413 if isinstance(path, str): 

414 path = {path: value} 

415 batch_size = batch_size or self.batch_size 

416 return await _run_coros_in_chunks( 

417 [self._pipe_file(k, v, **kwargs) for k, v in path.items()], 

418 batch_size=batch_size, 

419 nofiles=True, 

420 ) 

421 

422 async def _process_limits(self, url, start, end): 

423 """Helper for "Range"-based _cat_file""" 

424 size = None 

425 suff = False 

426 if start is not None and start < 0: 

427 # if start is negative and end None, end is the "suffix length" 

428 if end is None: 

429 end = -start 

430 start = "" 

431 suff = True 

432 else: 

433 size = size or (await self._info(url))["size"] 

434 start = size + start 

435 elif start is None: 

436 start = 0 

437 if not suff: 

438 if end is not None and end < 0: 

439 if start is not None: 

440 size = size or (await self._info(url))["size"] 

441 end = size + end 

442 elif end is None: 

443 end = "" 

444 if isinstance(end, numbers.Integral): 

445 end -= 1 # bytes range is inclusive 

446 return f"bytes={start}-{end}" 

447 

448 async def _cat_file(self, path, start=None, end=None, **kwargs): 

449 raise NotImplementedError 

450 

451 async def _cat( 

452 self, path, recursive=False, on_error="raise", batch_size=None, **kwargs 

453 ): 

454 paths = await self._expand_path(path, recursive=recursive) 

455 coros = [self._cat_file(path, **kwargs) for path in paths] 

456 batch_size = batch_size or self.batch_size 

457 out = await _run_coros_in_chunks( 

458 coros, batch_size=batch_size, nofiles=True, return_exceptions=True 

459 ) 

460 if on_error == "raise": 

461 ex = next(filter(is_exception, out), False) 

462 if ex: 

463 raise ex 

464 if ( 

465 len(paths) > 1 

466 or isinstance(path, list) 

467 or paths[0] != self._strip_protocol(path) 

468 ): 

469 return { 

470 k: v 

471 for k, v in zip(paths, out) 

472 if on_error != "omit" or not is_exception(v) 

473 } 

474 else: 

475 return out[0] 

476 

477 async def _cat_ranges( 

478 self, 

479 paths, 

480 starts, 

481 ends, 

482 max_gap=None, 

483 batch_size=None, 

484 on_error="return", 

485 **kwargs, 

486 ): 

487 """Get the contents of byte ranges from one or more files 

488 

489 Parameters 

490 ---------- 

491 paths: list 

492 A list of of filepaths on this filesystems 

493 starts, ends: int or list 

494 Bytes limits of the read. If using a single int, the same value will be 

495 used to read all the specified files. 

496 """ 

497 # TODO: on_error 

498 if max_gap is not None: 

499 # use utils.merge_offset_ranges 

500 raise NotImplementedError 

501 if not isinstance(paths, list): 

502 raise TypeError 

503 if not isinstance(starts, Iterable): 

504 starts = [starts] * len(paths) 

505 if not isinstance(ends, Iterable): 

506 ends = [ends] * len(paths) 

507 if len(starts) != len(paths) or len(ends) != len(paths): 

508 raise ValueError 

509 coros = [ 

510 self._cat_file(p, start=s, end=e, **kwargs) 

511 for p, s, e in zip(paths, starts, ends) 

512 ] 

513 batch_size = batch_size or self.batch_size 

514 return await _run_coros_in_chunks( 

515 coros, batch_size=batch_size, nofiles=True, return_exceptions=True 

516 ) 

517 

518 async def _put_file(self, lpath, rpath, mode="overwrite", **kwargs): 

519 raise NotImplementedError 

520 

521 async def _put( 

522 self, 

523 lpath, 

524 rpath, 

525 recursive=False, 

526 callback=DEFAULT_CALLBACK, 

527 batch_size=None, 

528 maxdepth=None, 

529 **kwargs, 

530 ): 

531 """Copy file(s) from local. 

532 

533 Copies a specific file or tree of files (if recursive=True). If rpath 

534 ends with a "/", it will be assumed to be a directory, and target files 

535 will go within. 

536 

537 The put_file method will be called concurrently on a batch of files. The 

538 batch_size option can configure the amount of futures that can be executed 

539 at the same time. If it is -1, then all the files will be uploaded concurrently. 

540 The default can be set for this instance by passing "batch_size" in the 

541 constructor, or for all instances by setting the "gather_batch_size" key 

542 in ``fsspec.config.conf``, falling back to 1/8th of the system limit . 

543 """ 

544 if isinstance(lpath, list) and isinstance(rpath, list): 

545 # No need to expand paths when both source and destination 

546 # are provided as lists 

547 rpaths = rpath 

548 lpaths = lpath 

549 else: 

550 source_is_str = isinstance(lpath, str) 

551 if source_is_str: 

552 lpath = make_path_posix(lpath) 

553 fs = LocalFileSystem() 

554 lpaths = fs.expand_path(lpath, recursive=recursive, maxdepth=maxdepth) 

555 if source_is_str and (not recursive or maxdepth is not None): 

556 # Non-recursive glob does not copy directories 

557 lpaths = [p for p in lpaths if not (trailing_sep(p) or fs.isdir(p))] 

558 if not lpaths: 

559 return 

560 

561 source_is_file = len(lpaths) == 1 

562 dest_is_dir = isinstance(rpath, str) and ( 

563 trailing_sep(rpath) or await self._isdir(rpath) 

564 ) 

565 

566 rpath = self._strip_protocol(rpath) 

567 exists = source_is_str and ( 

568 (has_magic(lpath) and source_is_file) 

569 or (not has_magic(lpath) and dest_is_dir and not trailing_sep(lpath)) 

570 ) 

571 rpaths = other_paths( 

572 lpaths, 

573 rpath, 

574 exists=exists, 

575 flatten=not source_is_str, 

576 ) 

577 

578 is_dir = {l: os.path.isdir(l) for l in lpaths} 

579 rdirs = [r for l, r in zip(lpaths, rpaths) if is_dir[l]] 

580 file_pairs = [(l, r) for l, r in zip(lpaths, rpaths) if not is_dir[l]] 

581 

582 await asyncio.gather(*[self._makedirs(d, exist_ok=True) for d in rdirs]) 

583 batch_size = batch_size or self.batch_size 

584 

585 coros = [] 

586 callback.set_size(len(file_pairs)) 

587 for lfile, rfile in file_pairs: 

588 put_file = callback.branch_coro(self._put_file) 

589 coros.append(put_file(lfile, rfile, **kwargs)) 

590 

591 return await _run_coros_in_chunks( 

592 coros, batch_size=batch_size, callback=callback 

593 ) 

594 

595 async def _get_file(self, rpath, lpath, **kwargs): 

596 raise NotImplementedError 

597 

598 async def _get( 

599 self, 

600 rpath, 

601 lpath, 

602 recursive=False, 

603 callback=DEFAULT_CALLBACK, 

604 maxdepth=None, 

605 **kwargs, 

606 ): 

607 """Copy file(s) to local. 

608 

609 Copies a specific file or tree of files (if recursive=True). If lpath 

610 ends with a "/", it will be assumed to be a directory, and target files 

611 will go within. Can submit a list of paths, which may be glob-patterns 

612 and will be expanded. 

613 

614 The get_file method will be called concurrently on a batch of files. The 

615 batch_size option can configure the amount of futures that can be executed 

616 at the same time. If it is -1, then all the files will be uploaded concurrently. 

617 The default can be set for this instance by passing "batch_size" in the 

618 constructor, or for all instances by setting the "gather_batch_size" key 

619 in ``fsspec.config.conf``, falling back to 1/8th of the system limit . 

620 """ 

621 if isinstance(lpath, list) and isinstance(rpath, list): 

622 # No need to expand paths when both source and destination 

623 # are provided as lists 

624 rpaths = rpath 

625 lpaths = lpath 

626 else: 

627 source_is_str = isinstance(rpath, str) 

628 # First check for rpath trailing slash as _strip_protocol removes it. 

629 source_not_trailing_sep = source_is_str and not trailing_sep(rpath) 

630 rpath = self._strip_protocol(rpath) 

631 rpaths = await self._expand_path( 

632 rpath, recursive=recursive, maxdepth=maxdepth 

633 ) 

634 if source_is_str and (not recursive or maxdepth is not None): 

635 # Non-recursive glob does not copy directories 

636 rpaths = [ 

637 p for p in rpaths if not (trailing_sep(p) or await self._isdir(p)) 

638 ] 

639 if not rpaths: 

640 return 

641 

642 lpath = make_path_posix(lpath) 

643 source_is_file = len(rpaths) == 1 

644 dest_is_dir = isinstance(lpath, str) and ( 

645 trailing_sep(lpath) or LocalFileSystem().isdir(lpath) 

646 ) 

647 

648 exists = source_is_str and ( 

649 (has_magic(rpath) and source_is_file) 

650 or (not has_magic(rpath) and dest_is_dir and source_not_trailing_sep) 

651 ) 

652 lpaths = other_paths( 

653 rpaths, 

654 lpath, 

655 exists=exists, 

656 flatten=not source_is_str, 

657 ) 

658 

659 [os.makedirs(os.path.dirname(lp), exist_ok=True) for lp in lpaths] 

660 batch_size = kwargs.pop("batch_size", self.batch_size) 

661 

662 coros = [] 

663 callback.set_size(len(lpaths)) 

664 for lpath, rpath in zip(lpaths, rpaths): 

665 get_file = callback.branch_coro(self._get_file) 

666 coros.append(get_file(rpath, lpath, **kwargs)) 

667 return await _run_coros_in_chunks( 

668 coros, batch_size=batch_size, callback=callback 

669 ) 

670 

671 async def _isfile(self, path): 

672 try: 

673 return (await self._info(path))["type"] == "file" 

674 except: # noqa: E722 

675 return False 

676 

677 async def _isdir(self, path): 

678 try: 

679 return (await self._info(path))["type"] == "directory" 

680 except OSError: 

681 return False 

682 

683 async def _size(self, path): 

684 return (await self._info(path)).get("size", None) 

685 

686 async def _sizes(self, paths, batch_size=None): 

687 batch_size = batch_size or self.batch_size 

688 return await _run_coros_in_chunks( 

689 [self._size(p) for p in paths], batch_size=batch_size 

690 ) 

691 

692 async def _exists(self, path, **kwargs): 

693 try: 

694 await self._info(path, **kwargs) 

695 return True 

696 except FileNotFoundError: 

697 return False 

698 

699 async def _info(self, path, **kwargs): 

700 raise NotImplementedError 

701 

702 async def _ls(self, path, detail=True, **kwargs): 

703 raise NotImplementedError 

704 

705 async def _walk(self, path, maxdepth=None, on_error="omit", **kwargs): 

706 if maxdepth is not None and maxdepth < 1: 

707 raise ValueError("maxdepth must be at least 1") 

708 

709 path = self._strip_protocol(path) 

710 full_dirs = {} 

711 dirs = {} 

712 files = {} 

713 

714 detail = kwargs.pop("detail", False) 

715 try: 

716 listing = await self._ls(path, detail=True, **kwargs) 

717 except (FileNotFoundError, OSError) as e: 

718 if on_error == "raise": 

719 raise 

720 elif callable(on_error): 

721 on_error(e) 

722 if detail: 

723 yield path, {}, {} 

724 else: 

725 yield path, [], [] 

726 return 

727 

728 for info in listing: 

729 # each info name must be at least [path]/part , but here 

730 # we check also for names like [path]/part/ 

731 pathname = info["name"].rstrip("/") 

732 name = pathname.rsplit("/", 1)[-1] 

733 if info["type"] == "directory" and pathname != path: 

734 # do not include "self" path 

735 full_dirs[name] = pathname 

736 dirs[name] = info 

737 elif pathname == path: 

738 # file-like with same name as give path 

739 files[""] = info 

740 else: 

741 files[name] = info 

742 

743 if detail: 

744 yield path, dirs, files 

745 else: 

746 yield path, list(dirs), list(files) 

747 

748 if maxdepth is not None: 

749 maxdepth -= 1 

750 if maxdepth < 1: 

751 return 

752 

753 for d in dirs: 

754 async for _ in self._walk( 

755 full_dirs[d], maxdepth=maxdepth, detail=detail, **kwargs 

756 ): 

757 yield _ 

758 

759 async def _glob(self, path, maxdepth=None, **kwargs): 

760 if maxdepth is not None and maxdepth < 1: 

761 raise ValueError("maxdepth must be at least 1") 

762 

763 import re 

764 

765 seps = (os.path.sep, os.path.altsep) if os.path.altsep else (os.path.sep,) 

766 ends_with_sep = path.endswith(seps) # _strip_protocol strips trailing slash 

767 path = self._strip_protocol(path) 

768 append_slash_to_dirname = ends_with_sep or path.endswith( 

769 tuple(sep + "**" for sep in seps) 

770 ) 

771 idx_star = path.find("*") if path.find("*") >= 0 else len(path) 

772 idx_qmark = path.find("?") if path.find("?") >= 0 else len(path) 

773 idx_brace = path.find("[") if path.find("[") >= 0 else len(path) 

774 

775 min_idx = min(idx_star, idx_qmark, idx_brace) 

776 

777 detail = kwargs.pop("detail", False) 

778 

779 if not has_magic(path): 

780 if await self._exists(path, **kwargs): 

781 if not detail: 

782 return [path] 

783 else: 

784 return {path: await self._info(path, **kwargs)} 

785 else: 

786 if not detail: 

787 return [] # glob of non-existent returns empty 

788 else: 

789 return {} 

790 elif "/" in path[:min_idx]: 

791 min_idx = path[:min_idx].rindex("/") 

792 root = path[: min_idx + 1] 

793 depth = path[min_idx + 1 :].count("/") + 1 

794 else: 

795 root = "" 

796 depth = path[min_idx + 1 :].count("/") + 1 

797 

798 if "**" in path: 

799 if maxdepth is not None: 

800 idx_double_stars = path.find("**") 

801 depth_double_stars = path[idx_double_stars:].count("/") + 1 

802 depth = depth - depth_double_stars + maxdepth 

803 else: 

804 depth = None 

805 

806 allpaths = await self._find( 

807 root, maxdepth=depth, withdirs=True, detail=True, **kwargs 

808 ) 

809 

810 pattern = glob_translate(path + ("/" if ends_with_sep else "")) 

811 pattern = re.compile(pattern) 

812 

813 out = { 

814 p: info 

815 for p, info in sorted(allpaths.items()) 

816 if pattern.match( 

817 p + "/" 

818 if append_slash_to_dirname and info["type"] == "directory" 

819 else p 

820 ) 

821 } 

822 

823 if detail: 

824 return out 

825 else: 

826 return list(out) 

827 

828 async def _du(self, path, total=True, maxdepth=None, **kwargs):