Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/fsspec/asyn.py: 26%
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
1import asyncio
2import asyncio.events
3import atexit
4import concurrent.futures
5import functools
6import inspect
7import io
8import logging
9import numbers
10import os
11import re
12import sys
13import threading
14from collections.abc import Iterable
15from glob import has_magic
16from typing import TYPE_CHECKING
18from .callbacks import DEFAULT_CALLBACK
19from .exceptions import FSTimeoutError
20from .implementations.local import LocalFileSystem, make_path_posix, trailing_sep
21from .spec import AbstractBufferedFile, AbstractFileSystem
22from .utils import check_contained, glob_translate, is_exception, other_paths
24private = re.compile("_[^_]")
25iothread = [None] # dedicated fsspec IO thread
26loop = [None] # global event loop for any non-async instance
27_lock = threading.Lock()
28get_running_loop = asyncio.get_running_loop
31def get_lock():
32 """Return the process-local threading lock."""
33 return _lock
36def reset_lock():
37 """Reset the global loop and lock after forking."""
38 global _lock
40 iothread[0] = None
41 loop[0] = None
42 _lock = threading.Lock()
45async def _runner(event, coro, result, timeout=None):
46 timeout = timeout if timeout else None # convert 0 or 0.0 to None
47 if timeout is not None:
48 coro = asyncio.wait_for(coro, timeout=timeout)
49 try:
50 result[0] = await coro
51 except Exception as ex:
52 result[0] = ex
53 finally:
54 event.set()
57def sync(loop, func, *args, timeout=None, **kwargs):
58 """
59 Make loop run coroutine until it returns. Runs in other thread
61 Examples
62 --------
63 >>> fsspec.asyn.sync(fsspec.asyn.get_loop(), func, *args,
64 timeout=timeout, **kwargs)
65 """
66 timeout = timeout if timeout else None # convert 0 or 0.0 to None
67 # NB: if the loop is not running *yet*, it is OK to submit work
68 # and we will wait for it
69 if loop is None or loop.is_closed():
70 raise RuntimeError("Loop is not running")
71 if iothread[0] is not None and (
72 loop is globals()["loop"][0]
73 or (getattr(loop, "_thread_id", 0) or 0) == iothread[0].ident
74 ):
75 if not iothread[0].is_alive():
76 raise RuntimeError("fsspec IO thread has already stopped.")
77 try:
78 loop0 = asyncio.events.get_running_loop()
79 if loop0 is loop:
80 raise NotImplementedError("Calling sync() from within a running loop")
81 except NotImplementedError:
82 raise
83 except RuntimeError:
84 pass
85 coro = func(*args, **kwargs)
86 result = [None]
87 event = threading.Event()
88 asyncio.run_coroutine_threadsafe(_runner(event, coro, result, timeout), loop)
89 while True:
90 # this loops allows thread to get interrupted
91 if event.wait(1):
92 break
93 if timeout is not None:
94 timeout -= 1
95 if timeout < 0:
96 raise FSTimeoutError
98 return_result = result[0]
99 # the traceback of a stored exception includes _runner's frame, which holds
100 # ``result``: empty it so that frame does not refer back to the exception
101 result[0] = None
102 try:
103 if isinstance(return_result, asyncio.TimeoutError):
104 # suppress asyncio.TimeoutError, raise FSTimeoutError
105 raise FSTimeoutError from return_result
106 elif isinstance(return_result, BaseException):
107 raise return_result
108 else:
109 return return_result
110 finally:
111 # the traceback also includes this frame, so don't keep the exception
112 # in a local either: that would be a reference cycle
113 del return_result
116_deferred_close_tasks = set()
117_deferred_close_lock = threading.Lock()
120def _on_loop_thread(loop):
121 """Returns True if the current thread is servicing the given event loop."""
122 if loop is None:
123 return False
124 try:
125 return asyncio.get_running_loop() is loop
126 except RuntimeError:
127 return False
130def _defer_task(
131 loop,
132 coro,
133 description="deferred task",
134 logger=None,
135 log_level=logging.WARNING,
136):
137 """Schedules a coroutine as a tracked background task on ``loop``.
139 Retains a strong reference in ``_deferred_close_tasks`` until completion to
140 prevent asyncio garbage collection from discarding pending tasks mid-flight,
141 and ensures unhandled task exceptions are retrieved and logged.
142 """
143 task = loop.create_task(coro)
144 with _deferred_close_lock:
145 _deferred_close_tasks.add(task)
147 def _on_done(t):
148 with _deferred_close_lock:
149 _deferred_close_tasks.discard(t)
150 if not t.cancelled():
151 exc = t.exception()
152 if exc:
153 log = logger or logging.getLogger("fsspec.asyn")
154 log.log(
155 log_level,
156 "%s failed during asynchronous execution: %s",
157 description,
158 exc,
159 exc_info=exc,
160 )
162 task.add_done_callback(_on_done)
163 return task
166def sync_teardown(
167 loop,
168 func_or_coro,
169 *args,
170 timeout=None,
171 description="teardown",
172 **kwargs,
173):
174 """Safely runs an async teardown coroutine on ``loop`` from synchronous context.
176 Schedules via :func:`asyncio.run_coroutine_threadsafe` or defers on the loop
177 thread to prevent deadlocks.
178 """
179 coro = func_or_coro(*args, **kwargs) if callable(func_or_coro) else func_or_coro
180 if not asyncio.iscoroutine(coro):
181 return
183 if sys.is_finalizing():
184 coro.close()
185 return
187 if loop is None or not loop.is_running() or loop.is_closed():
188 coro.close()
189 raise RuntimeError(f"Skipping {description}: no usable IO loop available.")
191 if _on_loop_thread(loop):
192 _defer_task(loop, coro, description=description, log_level=logging.ERROR)
193 return
195 try:
196 future = asyncio.run_coroutine_threadsafe(coro, loop)
197 except RuntimeError:
198 coro.close()
199 raise RuntimeError(f"Skipping {description}: event loop is closed.") from None
201 if timeout is not None and timeout <= 0:
202 return
204 try:
205 return future.result(timeout)
206 except concurrent.futures.TimeoutError:
207 raise FSTimeoutError(
208 f"{description} did not complete within {timeout}s."
209 ) from None
212def sync_wrapper(func, obj=None):
213 """Given a function, make so can be called in blocking contexts
215 Leave obj=None if defining within a class. Pass the instance if attaching
216 as an attribute of the instance.
217 """
219 @functools.wraps(func)
220 def wrapper(*args, **kwargs):
221 self = obj or args[0]
222 return sync(self.loop, func, *args, **kwargs)
224 return wrapper
227def async_gen_wrapper(func, obj=None):
228 """Given a async generator, make so can be called in blocking contexts"""
230 @functools.wraps(func)
231 def wrapper(*args, **kwargs):
232 self = obj or args[0]
233 gen = func(*args, **kwargs)
234 while True:
235 try:
236 yield sync(self.loop, gen.__anext__)
237 except StopAsyncIteration:
238 break
240 return wrapper
243def get_loop():
244 """Create or return the default fsspec IO loop
246 The loop will be running on a separate thread.
247 """
248 if loop[0] is None:
249 with get_lock():
250 # repeat the check just in case the loop got filled between the
251 # previous two calls from another thread
252 if loop[0] is None:
253 loop[0] = asyncio.new_event_loop()
254 th = threading.Thread(target=loop[0].run_forever, name="fsspecIO")
255 th.daemon = True
256 th.start()
257 iothread[0] = th
258 return loop[0]
261# How long to wait for the IO thread at exit before warning that it is stuck
262_STOP_IO_LOOP_WARN_SECONDS = 1
265def _stop_io_loop():
266 """Stop the fsspec IO loop, if one was started, and wait for its thread.
268 The IO thread is a daemon, so nothing else stops it before interpreter
269 shutdown. It must not outlive ``Py_Finalize``: a callback firing on it
270 afterwards (for example a uvloop timer) re-enters the finalized interpreter
271 and crashes the process (#2241). atexit hooks run before finalization,
272 while the thread can still take the GIL to process the stop.
274 The stop takes effect once the running callback returns, so the wait only
275 lasts if a callback blocks the loop. That case is logged, then waited out:
276 letting the thread survive finalization would reintroduce the crash.
277 """
278 if loop[0] is None or iothread[0] is None or not iothread[0].is_alive():
279 return
280 try:
281 loop[0].call_soon_threadsafe(loop[0].stop)
282 except RuntimeError:
283 # the loop was closed by its user
284 return
285 iothread[0].join(timeout=_STOP_IO_LOOP_WARN_SECONDS)
286 if iothread[0].is_alive():
287 logging.getLogger("fsspec.asyn").warning(
288 "fsspec IO loop is still running %ss after interpreter exit began; "
289 "a callback is blocking it. Waiting for it to finish.",
290 _STOP_IO_LOOP_WARN_SECONDS,
291 )
292 iothread[0].join()
295# Registered at import, so it runs after any atexit hook registered later
296# that may still use the loop. A forked child inherits this registration;
297# the fork guard below clears ``loop``/``iothread`` there, so in the child it
298# only stops a loop the child started itself, never the parent's thread.
299atexit.register(_stop_io_loop)
302def reset_after_fork():
303 reset_lock()
306if hasattr(os, "register_at_fork"):
307 # should be posix; this will do nothing for spawn or forkserver subprocesses
308 os.register_at_fork(after_in_child=reset_after_fork)
311if TYPE_CHECKING:
312 import resource
314 ResourceError = resource.error
315else:
316 try:
317 import resource
318 except ImportError:
319 resource = None
320 ResourceError = OSError
321 else:
322 ResourceError = getattr(resource, "error", OSError)
324_DEFAULT_BATCH_SIZE = 128
325_NOFILES_DEFAULT_BATCH_SIZE = 1280
328def _get_batch_size(nofiles=False):
329 from fsspec.config import conf
331 if nofiles:
332 if "nofiles_gather_batch_size" in conf:
333 return conf["nofiles_gather_batch_size"]
334 else:
335 if "gather_batch_size" in conf:
336 return conf["gather_batch_size"]
337 if nofiles:
338 return _NOFILES_DEFAULT_BATCH_SIZE
339 if resource is None:
340 return _DEFAULT_BATCH_SIZE
342 try:
343 soft_limit, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
344 except (ImportError, ValueError, ResourceError):
345 return _DEFAULT_BATCH_SIZE
347 if soft_limit == resource.RLIM_INFINITY:
348 return -1
349 else:
350 return soft_limit // 8
353def running_async() -> bool:
354 """Being executed by an event loop?"""
355 try:
356 asyncio.get_running_loop()
357 return True
358 except RuntimeError:
359 return False
362async def _run_coros_in_chunks(
363 coros,
364 batch_size=None,
365 callback=DEFAULT_CALLBACK,
366 timeout=None,
367 return_exceptions=False,
368 nofiles=False,
369):
370 """Run the given coroutines in chunks.
372 Parameters
373 ----------
374 coros: list of coroutines to run
375 batch_size: int or None
376 Number of coroutines to submit/wait on simultaneously.
377 If -1, then it will not be any throttling. If
378 None, it will be inferred from _get_batch_size()
379 callback: fsspec.callbacks.Callback instance
380 Gets a relative_update when each coroutine completes
381 timeout: number or None
382 If given, each coroutine times out after this time. Note that, since
383 there are multiple batches, the total run time of this function will in
384 general be longer
385 return_exceptions: bool
386 Same meaning as in asyncio.gather
387 nofiles: bool
388 If inferring the batch_size, does this operation involve local files?
389 If yes, you normally expect smaller batches.
390 """
392 if batch_size is None:
393 batch_size = _get_batch_size(nofiles=nofiles)
395 if batch_size == -1:
396 batch_size = len(coros)
397 elif batch_size <= 0:
398 raise ValueError
400 async def _run_coro(coro, i):
401 try:
402 return await asyncio.wait_for(coro, timeout=timeout), i
403 except Exception as e:
404 if not return_exceptions:
405 raise
406 return e, i
407 finally:
408 callback.relative_update(1)
410 i = 0
411 n = len(coros)
412 results = [None] * n
413 pending = set()
415 while pending or i < n:
416 while len(pending) < batch_size and i < n:
417 pending.add(asyncio.ensure_future(_run_coro(coros[i], i)))
418 i += 1
420 if not pending:
421 break
423 done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
424 first_exc = None
425 while done:
426 task = done.pop()
427 try:
428 result, k = await task
429 results[k] = result
430 except Exception as exc:
431 if first_exc is None:
432 first_exc = exc
434 if first_exc is not None:
435 for task in pending:
436 task.cancel()
437 if pending:
438 await asyncio.gather(*pending, return_exceptions=True)
439 try:
440 raise first_exc
441 finally:
442 # the traceback includes this frame, so it must not refer back
443 # to the exception: directly or through the finished task
444 del first_exc, task
446 return results
449# these methods should be implemented as async by any async-able backend
450async_methods = [
451 "_ls",
452 "_cat_file",
453 "_get_file",
454 "_put_file",
455 "_rm_file",
456 "_cp_file",
457 "_pipe_file",
458 "_expand_path",
459 "_info",
460 "_isfile",
461 "_isdir",
462 "_exists",
463 "_walk",
464 "_glob",
465 "_find",
466 "_du",
467 "_size",
468 "_mkdir",
469 "_makedirs",
470]
473class AsyncFileSystem(AbstractFileSystem):
474 """Async file operations, default implementations
476 Passes bulk operations to asyncio.gather for concurrent operation.
478 Implementations that have concurrent batch operations and/or async methods
479 should inherit from this class instead of AbstractFileSystem. Docstrings are
480 copied from the un-underscored method in AbstractFileSystem, if not given.
481 """
483 # note that methods do not have docstring here; they will be copied
484 # for _* methods and inferred for overridden methods.
486 async_impl = True
487 mirror_sync_methods = True
488 disable_throttling = False
490 def __init__(self, *args, asynchronous=False, loop=None, batch_size=None, **kwargs):
491 self.asynchronous = asynchronous
492 self._pid = os.getpid()
493 if not asynchronous:
494 self._loop = loop or get_loop()
495 else:
496 self._loop = None
497 self.batch_size = batch_size
498 super().__init__(*args, **kwargs)
500 @property
501 def loop(self):
502 if self._pid != os.getpid():
503 raise RuntimeError("This class is not fork-safe")
504 return self._loop
506 async def _rm_file(self, path, **kwargs):
507 if (
508 inspect.iscoroutinefunction(self._rm)
509 and type(self)._rm is not AsyncFileSystem._rm
510 ):
511 return await self._rm(path, recursive=False, batch_size=1, **kwargs)
512 raise NotImplementedError
514 async def _rm(
515 self, path, recursive=False, batch_size=None, maxdepth=None, **kwargs
516 ):
517 # TODO: implement on_error
518 batch_size = batch_size or self.batch_size
519 path = await self._expand_path(path, recursive=recursive, maxdepth=maxdepth)
520 return await _run_coros_in_chunks(
521 [self._rm_file(p, **kwargs) for p in reversed(path)],
522 batch_size=batch_size,
523 nofiles=True,
524 )
526 async def _cp_file(self, path1, path2, **kwargs):
527 raise NotImplementedError
529 async def _mv_file(self, path1, path2):
530 await self._cp_file(path1, path2)
531 await self._rm_file(path1)
533 async def _copy(
534 self,
535 path1,
536 path2,
537 recursive=False,
538 on_error=None,
539 maxdepth=None,
540 batch_size=None,
541 **kwargs,
542 ):
543 if on_error is None and recursive:
544 on_error = "ignore"
545 elif on_error is None:
546 on_error = "raise"
548 if isinstance(path1, list) and isinstance(path2, list):
549 # No need to expand paths when both source and destination
550 # are provided as lists
551 paths1 = path1
552 paths2 = path2
553 else:
554 source_is_str = isinstance(path1, str)
555 paths1 = await self._expand_path(
556 path1, maxdepth=maxdepth, recursive=recursive
557 )
558 if source_is_str and (not recursive or maxdepth is not None):
559 # Non-recursive glob does not copy directories
560 paths1 = [
561 p for p in paths1 if not (trailing_sep(p) or await self._isdir(p))
562 ]
563 if not paths1:
564 return
566 source_is_file = len(paths1) == 1
567 dest_is_dir = isinstance(path2, str) and (
568 trailing_sep(path2) or await self._isdir(path2)
569 )
571 exists = source_is_str and (
572 (has_magic(path1) and source_is_file)
573 or (not has_magic(path1) and dest_is_dir and not trailing_sep(path1))
574 )
575 paths2 = other_paths(
576 paths1,
577 path2,
578 exists=exists,
579 flatten=not source_is_str,
580 )
582 batch_size = batch_size or self.batch_size
583 coros = [self._cp_file(p1, p2, **kwargs) for p1, p2 in zip(paths1, paths2)]
584 result = await _run_coros_in_chunks(
585 coros, batch_size=batch_size, return_exceptions=True, nofiles=True
586 )
588 for ex in filter(is_exception, result):
589 if on_error == "ignore" and isinstance(ex, FileNotFoundError):
590 continue
591 try:
592 raise ex
593 finally:
594 del ex, result
596 async def _pipe_file(self, path, value, mode="overwrite", **kwargs):
597 raise NotImplementedError
599 async def _pipe(self, path, value=None, batch_size=None, **kwargs):
600 if isinstance(path, str):
601 path = {path: value}
602 batch_size = batch_size or self.batch_size
603 return await _run_coros_in_chunks(
604 [self._pipe_file(k, v, **kwargs) for k, v in path.items()],
605 batch_size=batch_size,
606 nofiles=True,
607 )
609 async def _process_limits(self, url, start, end):
610 """Helper for "Range"-based _cat_file"""
611 size = None
612 suff = False
613 if start is not None and start < 0:
614 # if start is negative and end None, end is the "suffix length"
615 if end is None:
616 end = -start
617 start = ""
618 suff = True
619 else:
620 size = size or (await self._info(url))["size"]
621 start = size + start
622 elif start is None:
623 start = 0
624 if not suff:
625 if end is not None and end < 0:
626 if start is not None:
627 size = size or (await self._info(url))["size"]
628 end = size + end
629 elif end is None:
630 end = ""
631 if isinstance(end, numbers.Integral):
632 end -= 1 # bytes range is inclusive
633 return f"bytes={start}-{end}"
635 async def _cat_file(self, path, start=None, end=None, **kwargs):
636 raise NotImplementedError
638 async def _cat(
639 self, path, recursive=False, on_error="raise", batch_size=None, **kwargs
640 ):
641 paths = await self._expand_path(path, recursive=recursive)
642 coros = [self._cat_file(path, **kwargs) for path in paths]
643 batch_size = batch_size or self.batch_size
644 out = await _run_coros_in_chunks(
645 coros, batch_size=batch_size, nofiles=True, return_exceptions=True
646 )
647 if on_error == "raise":
648 ex = next(filter(is_exception, out), False)
649 if ex:
650 try:
651 raise ex
652 finally:
653 del ex, out
654 if (
655 len(paths) > 1
656 or isinstance(path, list)
657 or paths[0] != self._strip_protocol(path)
658 ):
659 return {
660 k: v
661 for k, v in zip(paths, out)
662 if on_error != "omit" or not is_exception(v)
663 }
664 else:
665 return out[0]
667 async def _cat_ranges(
668 self,
669 paths,
670 starts,
671 ends,
672 max_gap=None,
673 batch_size=None,
674 on_error="return",
675 **kwargs,
676 ):
677 """Get the contents of byte ranges from one or more files
679 Parameters
680 ----------
681 paths: list
682 A list of of filepaths on this filesystems
683 starts, ends: int or list
684 Bytes limits of the read. If using a single int, the same value will be
685 used to read all the specified files.
686 on_error: "return" or "raise"
687 If "return" (default), any per-range exception is placed in the output
688 list at the corresponding position. Otherwise the first such exception
689 is raised. Matches ``AbstractFileSystem.cat_ranges``.
690 """
691 if max_gap is not None:
692 # use utils.merge_offset_ranges
693 raise NotImplementedError
694 if not isinstance(paths, list):
695 raise TypeError
696 if not isinstance(starts, Iterable):
697 starts = [starts] * len(paths)
698 if not isinstance(ends, Iterable):
699 ends = [ends] * len(paths)
700 if len(starts) != len(paths) or len(ends) != len(paths):
701 raise ValueError
702 coros = [
703 self._cat_file(p, start=s, end=e, **kwargs)
704 for p, s, e in zip(paths, starts, ends)
705 ]
706 batch_size = batch_size or self.batch_size
707 out = await _run_coros_in_chunks(
708 coros, batch_size=batch_size, nofiles=True, return_exceptions=True
709 )
710 if on_error != "return":
711 ex = next(filter(is_exception, out), None)
712 if ex is not None:
713 try:
714 raise ex
715 finally:
716 del ex, out
717 return out
719 async def _put_file(self, lpath, rpath, mode="overwrite", **kwargs):
720 raise NotImplementedError
722 async def _put(
723 self,
724 lpath,
725 rpath,
726 recursive=False,
727 callback=DEFAULT_CALLBACK,
728 batch_size=None,
729 maxdepth=None,
730 **kwargs,
731 ):
732 """Copy file(s) from local.
734 Copies a specific file or tree of files (if recursive=True). If rpath
735 ends with a "/", it will be assumed to be a directory, and target files
736 will go within.
738 The put_file method will be called concurrently on a batch of files. The
739 batch_size option can configure the amount of futures that can be executed
740 at the same time. If it is -1, then all the files will be uploaded concurrently.
741 The default can be set for this instance by passing "batch_size" in the
742 constructor, or for all instances by setting the "gather_batch_size" key
743 in ``fsspec.config.conf``, falling back to 1/8th of the system limit .
744 """
745 if isinstance(lpath, list) and isinstance(rpath, list):
746 # No need to expand paths when both source and destination
747 # are provided as lists
748 rpaths = rpath
749 lpaths = lpath
750 else:
751 source_is_str = isinstance(lpath, str)
752 if source_is_str:
753 lpath = make_path_posix(lpath)
754 fs = LocalFileSystem()
755 lpaths = fs.expand_path(lpath, recursive=recursive, maxdepth=maxdepth)
756 if source_is_str and (not recursive or maxdepth is not None):
757 # Non-recursive glob does not copy directories
758 lpaths = [p for p in lpaths if not (trailing_sep(p) or fs.isdir(p))]
759 if not lpaths:
760 return
762 source_is_file = len(lpaths) == 1
763 dest_is_dir = isinstance(rpath, str) and (
764 trailing_sep(rpath) or await self._isdir(rpath)
765 )
767 rpath = self._strip_protocol(rpath)
768 exists = source_is_str and (
769 (has_magic(lpath) and source_is_file)
770 or (not has_magic(lpath) and dest_is_dir and not trailing_sep(lpath))
771 )
772 rpaths = other_paths(
773 lpaths,
774 rpath,
775 exists=exists,
776 flatten=not source_is_str,
777 )
779 is_dir = {l: os.path.isdir(l) for l in lpaths}
780 rdirs = [r for l, r in zip(lpaths, rpaths) if is_dir[l]]
781 file_pairs = [(l, r) for l, r in zip(lpaths, rpaths) if not is_dir[l]]
783 await asyncio.gather(*[self._makedirs(d, exist_ok=True) for d in rdirs])
784 batch_size = batch_size or self.batch_size
786 coros = []
787 callback.set_size(len(file_pairs))
788 for lfile, rfile in file_pairs:
789 put_file = callback.branch_coro(self._put_file)
790 coros.append(put_file(lfile, rfile, **kwargs))
792 return await _run_coros_in_chunks(
793 coros, batch_size=batch_size, callback=callback
794 )
796 async def _get_file(self, rpath, lpath, **kwargs):
797 raise NotImplementedError
799 async def _get(
800 self,
801 rpath,
802 lpath,
803 recursive=False,
804 callback=DEFAULT_CALLBACK,
805 maxdepth=None,
806 **kwargs,
807 ):
808 """Copy file(s) to local.
810 Copies a specific file or tree of files (if recursive=True). If lpath
811 ends with a "/", it will be assumed to be a directory, and target files
812 will go within. Can submit a list of paths, which may be glob-patterns
813 and will be expanded.
815 The get_file method will be called concurrently on a batch of files. The
816 batch_size option can configure the amount of futures that can be executed
817 at the same time. If it is -1, then all the files will be uploaded concurrently.
818 The default can be set for this instance by passing "batch_size" in the
819 constructor, or for all instances by setting the "gather_batch_size" key
820 in ``fsspec.config.conf``, falling back to 1/8th of the system limit .
821 """
822 if isinstance(lpath, list) and isinstance(rpath, list):
823 # No need to expand paths when both source and destination
824 # are provided as lists
825 rpaths = rpath
826 lpaths = lpath
827 else:
828 source_is_str = isinstance(rpath, str)
829 # First check for rpath trailing slash as _strip_protocol removes it.
830 source_not_trailing_sep = source_is_str and not trailing_sep(rpath)
831 rpath = self._strip_protocol(rpath)
832 rpaths = await self._expand_path(
833 rpath, recursive=recursive, maxdepth=maxdepth
834 )
835 if source_is_str and (not recursive or maxdepth is not None):
836 # Non-recursive glob does not copy directories
837 rpaths = [
838 p for p in rpaths if not (trailing_sep(p) or await self._isdir(p))
839 ]
840 if not rpaths:
841 return
843 lpath = make_path_posix(lpath)
844 source_is_file = len(rpaths) == 1
845 dest_is_dir = isinstance(lpath, str) and (
846 trailing_sep(lpath) or LocalFileSystem().isdir(lpath)
847 )
849 exists = source_is_str and (
850 (has_magic(rpath) and source_is_file)
851 or (not has_magic(rpath) and dest_is_dir and source_not_trailing_sep)
852 )
853 lpaths = other_paths(
854 rpaths,
855 lpath,
856 exists=exists,
857 flatten=not source_is_str,
858 )
859 if isinstance(lpath, str):
860 # The names came from the source listing; ".." in one of them
861 # would otherwise place the copy above the destination. When
862 # lpath is a list the caller named every destination itself.
863 check_contained(lpath, lpaths)
865 for lp in lpaths:
866 if parent := os.path.dirname(lp):
867 os.makedirs(parent, exist_ok=True)
868 batch_size = kwargs.pop("batch_size", self.batch_size)
870 coros = []
871 callback.set_size(len(lpaths))
872 for lpath, rpath in zip(lpaths, rpaths):
873 get_file = callback.branch_coro(self._get_file)
874 coros.append(get_file(rpath, lpath, **kwargs))
875 return await _run_coros_in_chunks(
876 coros, batch_size=batch_size, callback=callback
877 )
879 async def _isfile(self, path):
880 try:
881 return (await self._info(path))["type"] == "file"
882 except Exception:
883 return False
885 async def _isdir(self, path):
886 try:
887 return (await self._info(path))["type"] == "directory"
888 except OSError:
889 return False
891 async def _size(self, path):
892 return (await self._info(path)).get("size", None)
894 async def _sizes(self, paths, batch_size=None):
895 batch_size = batch_size or self.batch_size
896 return await _run_coros_in_chunks(
897 [self._size(p) for p in paths], batch_size=batch_size
898 )
900 async def _exists(self, path, **kwargs):
901 try:
902 await self._info(path, **kwargs)
903 return True
904 except FileNotFoundError:
905 return False
907 async def _info(self, path, **kwargs):
908 raise NotImplementedError
910 async def _ls(self, path, detail=True, **kwargs):
911 raise NotImplementedError
913 async def _walk(self, path, maxdepth=None, topdown=True, on_error="omit", **kwargs):
914 if maxdepth is not None and maxdepth < 1:
915 raise ValueError("maxdepth must be at least 1")
917 path = self._strip_protocol(path)
918 full_dirs = {}
919 dirs = {}
920 files = {}
922 detail = kwargs.pop("detail", False)
923 try:
924 listing = await self._ls(path, detail=True, **kwargs)
925 except (FileNotFoundError, OSError) as e:
926 if on_error == "raise":
927 raise
928 elif callable(on_error):
929 on_error(e)
930 if detail:
931 yield path, {}, {}
932 else:
933 yield path, [], []
934 return
936 for info in listing:
937 # each info name must be at least [path]/part , but here
938 # we check also for names like [path]/part/
939 pathname = info["name"].rstrip("/")
940 name = pathname.rsplit("/", 1)[-1]
941 if info["type"] == "directory" and pathname != path:
942 # do not include "self" path
943 full_dirs[name] = pathname
944 dirs[name] = info
945 elif pathname == path:
946 # file-like with same name as give path
947 files[""] = info
948 else:
949 files[name] = info
951 if not detail:
952 dirs = list(dirs)
953 files = list(files)
955 if topdown:
956 # Yield before recursion if walking top down
957 yield path, dirs, files
959 if maxdepth is not None:
960 maxdepth -= 1
961 if maxdepth < 1:
962 if not topdown:
963 yield path, dirs, files
964 return
966 for d in dirs:
967 async for _ in self._walk(
968 full_dirs[d],
969 maxdepth=maxdepth,
970 detail=detail,
971 topdown=topdown,
972 on_error=on_error,
973 **kwargs,
974 ):
975 yield _
977 if not topdown:
978 # Yield after recursion if walking bottom up
979 yield path, dirs, files
981 async def _glob(self, path, maxdepth=None, **kwargs):
982 if maxdepth is not None and maxdepth < 1:
983 raise ValueError("maxdepth must be at least 1")
985 import re
987 seps = (os.path.sep, os.path.altsep) if os.path.altsep else (os.path.sep,)
988 ends_with_sep = path.endswith(seps) # _strip_protocol strips trailing slash
989 path = self._strip_protocol(path)
990 append_slash_to_dirname = ends_with_sep or path.endswith(
991 tuple(sep + "**" for sep in seps)
992 )
993 idx_star = path.find("*") if path.find("*") >= 0 else len(path)
994 idx_qmark = path.find("?") if path.find("?") >= 0 else len(path)
995 idx_brace = path.find("[") if path.find("[") >= 0 else len(path)
997 min_idx = min(idx_star, idx_qmark, idx_brace)
999 detail = kwargs.pop("detail", False)
1000 withdirs = kwargs.pop("withdirs", True)
1002 if not has_magic(path):
1003 if await self._exists(path, **kwargs):
1004 if not detail:
1005 return [path]
1006 else:
1007 return {path: await self._info(path, **kwargs)}
1008 else:
1009 if not detail:
1010 return [] # glob of non-existent returns empty
1011 else:
1012 return {}
1013 elif "/" in path[:min_idx]:
1014 first_wildcard_idx = min_idx
1015 min_idx = path[:min_idx].rindex("/")
1016 root = path[
1017 : min_idx + 1
1018 ] # everything up to the last / before the first wildcard
1019 prefix = path[
1020 min_idx + 1 : first_wildcard_idx
1021 ] # stem between last "/" and first wildcard
1022 depth = path[min_idx + 1 :].count("/") + 1
1023 else:
1024 root = ""
1025 prefix = path[:min_idx] # stem up to the first wildcard
1026 depth = path[min_idx + 1 :].count("/") + 1
1028 if "**" in path:
1029 if maxdepth is not None:
1030 idx_double_stars = path.find("**")
1031 depth_double_stars = path[idx_double_stars:].count("/") + 1
1032 depth = depth - depth_double_stars + maxdepth
1033 else:
1034 depth = None
1036 # Pass the filename stem as prefix= so backends that support it such as
1037 # gcsfs, s3fs and adlfs can filter server-side up to the first wildcard.
1038 if prefix:
1039 kwargs["prefix"] = prefix
1040 allpaths = await self._find(
1041 root, maxdepth=depth, withdirs=withdirs, detail=True, **kwargs
1042 )
1044 pattern = glob_translate(path + ("/" if ends_with_sep else ""))
1045 pattern = re.compile(pattern)
1047 out = {
1048 p: info
1049 for p, info in sorted(allpaths.items())
1050 if pattern.match(
1051 p + "/"
1052 if append_slash_to_dirname and info["type"] == "directory"
1053 else p
1054 )
1055 }
1057 if detail:
1058 return out
1059 else:
1060 return list(out)
1062 async def _du(self, path, total=True, maxdepth=None, **kwargs):
1063 sizes = {}
1064 # async for?
1065 for f in await self._find(path, maxdepth=maxdepth, **kwargs):
1066 info = await self._info(f)
1067 sizes[info["name"]] = info["size"]
1068 if total:
1069 return sum(sizes.values())
1070 else:
1071 return sizes
1073 async def _find(self, path, maxdepth=None, withdirs=False, **kwargs):
1074 path = self._strip_protocol(path)
1075 out = {}
1076 detail = kwargs.pop("detail", False)
1078 # Add the root directory if withdirs is requested
1079 # This is needed for posix glob compliance
1080 if withdirs and path != "" and await self._isdir(path):
1081 out[path] = await self._info(path)
1083 # async for?
1084 async for _, dirs, files in self._walk(path, maxdepth, detail=True, **kwargs):
1085 if withdirs:
1086 files.update(dirs)
1087 out.update({info["name"]: info for name, info in files.items()})
1088 if not out and (await self._isfile(path)):
1089 # walk works on directories, but find should also return [path]
1090 # when path happens to be a file
1091 out[path] = {}
1092 names = sorted(out)
1093 if not detail:
1094 return names
1095 else:
1096 return {name: out[name] for name in names}
1098 async def _expand_path(
1099 self, path, recursive=False, maxdepth=None, assume_literal=False
1100 ):
1101 if maxdepth is not None and maxdepth < 1:
1102 raise ValueError("maxdepth must be at least 1")
1104 if isinstance(path, str):
1105 out = await self._expand_path([path], recursive, maxdepth)
1106 else:
1107 out = set()
1108 path = [self._strip_protocol(p) for p in path]
1109 for p in path: # can gather here
1110 if not assume_literal and has_magic(p):
1111 bit = set(await self._glob(p, maxdepth=maxdepth))
1112 out |= bit
1113 if recursive:
1114 # glob call above expanded one depth so if maxdepth is defined
1115 # then decrement it in expand_path call below. If it is zero
1116 # after decrementing then avoid expand_path call.
1117 if maxdepth is not None and maxdepth <= 1:
1118 continue
1119 out |= set(
1120 await self._expand_path(
1121 list(bit),
1122 recursive=recursive,
1123 maxdepth=maxdepth - 1 if maxdepth is not None else None,
1124 assume_literal=True,
1125 )
1126 )
1127 continue
1128 elif recursive:
1129 rec = set(await self._find(p, maxdepth=maxdepth, withdirs=True))
1130 out |= rec
1131 if p not in out and (recursive is False or (await self._exists(p))):
1132 # should only check once, for the root
1133 out.add(p)
1134 if not out:
1135 raise FileNotFoundError(path)
1136 return sorted(out)
1138 async def _mkdir(self, path, create_parents=True, **kwargs):
1139 pass # not necessary to implement, may not have directories
1141 async def _makedirs(self, path, exist_ok=False):
1142 pass # not necessary to implement, may not have directories
1144 async def open_async(self, path, mode="rb", **kwargs):
1145 if "b" not in mode or kwargs.get("compression"):
1146 raise ValueError
1147 raise NotImplementedError
1150def mirror_sync_methods(obj):
1151 """Populate sync and async methods for obj
1153 For each method will create a sync version if the name refers to an async method
1154 (coroutine) and there is no override in the child class; will create an async
1155 method for the corresponding sync method if there is no implementation.
1157 Uses the methods specified in
1158 - async_methods: the set that an implementation is expected to provide
1159 - default_async_methods: that can be derived from their sync version in
1160 AbstractFileSystem
1161 - AsyncFileSystem: async-specific default coroutines
1162 """
1163 from fsspec import AbstractFileSystem
1165 for method in set(async_methods + dir(AsyncFileSystem)):
1166 if not method.startswith("_"):
1167 continue
1168 smethod = method[1:]
1169 if private.match(method):
1170 isco = inspect.iscoroutinefunction(getattr(obj, method, None))
1171 unsync = getattr(getattr(obj, smethod, False), "__func__", None)
1172 is_default = unsync is getattr(AbstractFileSystem, smethod, "")
1173 if isco and is_default:
1174 mth = sync_wrapper(getattr(obj, method), obj=obj)
1175 elif inspect.isasyncgenfunction(getattr(obj, method, None)) and is_default:
1176 mth = async_gen_wrapper(getattr(obj, method), obj=obj)
1177 else:
1178 continue
1179 setattr(obj, smethod, mth)
1180 if not mth.__doc__:
1181 mth.__doc__ = getattr(
1182 getattr(AbstractFileSystem, smethod, None), "__doc__", ""
1183 )
1186class FSSpecCoroutineCancel(Exception):
1187 pass
1190def _dump_running_tasks(
1191 printout=True, cancel=True, exc=FSSpecCoroutineCancel, with_task=False
1192):
1193 import traceback
1195 tasks = [t for t in asyncio.tasks.all_tasks(loop[0]) if not t.done()]
1196 if printout:
1197 [task.print_stack() for task in tasks]
1198 out = [
1199 {
1200 "locals": task._coro.cr_frame.f_locals,
1201 "file": task._coro.cr_frame.f_code.co_filename,
1202 "firstline": task._coro.cr_frame.f_code.co_firstlineno,
1203 "linelo": task._coro.cr_frame.f_lineno,
1204 "stack": traceback.format_stack(task._coro.cr_frame),
1205 "task": task if with_task else None,
1206 }
1207 for task in tasks
1208 ]
1209 if cancel:
1210 for t in tasks:
1211 cbs = t._callbacks
1212 t.cancel()
1213 asyncio.futures.Future.set_exception(t, exc)
1214 asyncio.futures.Future.cancel(t)
1215 [cb[0](t) for cb in cbs] # cancels any dependent concurrent.futures
1216 try:
1217 t._coro.throw(exc) # exits coro, unless explicitly handled
1218 except exc:
1219 pass
1220 return out
1223class AbstractAsyncStreamedFile(AbstractBufferedFile):
1224 # no read buffering, and always auto-commit
1225 # TODO: readahead might still be useful here, but needs async version
1227 async def read(self, length=-1):
1228 """
1229 Return data from cache, or fetch pieces as necessary
1231 Parameters
1232 ----------
1233 length: int (-1)
1234 Number of bytes to read; if <0, all remaining bytes.
1235 """
1236 length = -1 if length is None else int(length)
1237 if self.mode != "rb":
1238 raise ValueError("File not in read mode")
1239 if length < 0:
1240 length = self.size - self.loc
1241 if self.closed:
1242 raise ValueError("I/O operation on closed file.")
1243 if length == 0:
1244 # don't even bother calling fetch
1245 return b""
1246 out = await self._fetch_range(self.loc, self.loc + length)
1247 self.loc += len(out)
1248 return out
1250 async def write(self, data):
1251 """
1252 Write data to buffer.
1254 Buffer only sent on flush() or if buffer is greater than
1255 or equal to blocksize.
1257 Parameters
1258 ----------
1259 data: bytes
1260 Set of bytes to be written.
1261 """
1262 if self.mode not in {"wb", "ab"}:
1263 raise ValueError("File not in write mode")
1264 if self.closed:
1265 raise ValueError("I/O operation on closed file.")
1266 if self.forced:
1267 raise ValueError("This file has been force-flushed, can only close")
1268 out = self.buffer.write(data)
1269 self.loc += out
1270 if self.buffer.tell() >= self.blocksize:
1271 await self.flush()
1272 return out
1274 async def close(self):
1275 """Close file
1277 Finalizes writes, discards cache
1278 """
1279 if getattr(self, "_unclosable", False):
1280 return
1281 if self.closed:
1282 return
1283 if self.mode == "rb":
1284 self.cache = None
1285 else:
1286 if not self.forced:
1287 await self.flush(force=True)
1289 if self.fs is not None:
1290 self.fs.invalidate_cache(self.path)
1291 self.fs.invalidate_cache(self.fs._parent(self.path))
1293 self.closed = True
1295 async def flush(self, force=False):
1296 if self.closed:
1297 raise ValueError("Flush on closed file")
1298 if force and self.forced:
1299 raise ValueError("Force flush cannot be called more than once")
1300 if force:
1301 self.forced = True
1303 if self.mode not in {"wb", "ab"}:
1304 # no-op to flush on read-mode
1305 return
1307 if not force and self.buffer.tell() < self.blocksize:
1308 # Defer write on small block
1309 return
1311 if self.offset is None:
1312 # Initialize a multipart upload
1313 self.offset = 0
1314 try:
1315 await self._initiate_upload()
1316 except:
1317 self.closed = True
1318 raise
1320 if await self._upload_chunk(final=force) is not False:
1321 self.offset += self.buffer.seek(0, 2)
1322 self.buffer = io.BytesIO()
1324 async def __aenter__(self):
1325 return self
1327 async def __aexit__(self, exc_type, exc_val, exc_tb):
1328 await self.close()
1330 async def _fetch_range(self, start, end):
1331 raise NotImplementedError
1333 async def _initiate_upload(self):
1334 pass
1336 async def _upload_chunk(self, final=False):
1337 raise NotImplementedError