Spaces:
Runtime error
Runtime error
File size: 1,982 Bytes
9cddcfd |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
from importlib.metadata import entry_points
from . import _version, caching
from .callbacks import Callback
from .compression import available_compressions
from .core import get_fs_token_paths, open, open_files, open_local
from .exceptions import FSTimeoutError
from .mapping import FSMap, get_mapper
from .registry import (
available_protocols,
filesystem,
get_filesystem_class,
register_implementation,
registry,
)
from .spec import AbstractFileSystem
__version__ = _version.get_versions()["version"]
__all__ = [
"AbstractFileSystem",
"FSTimeoutError",
"FSMap",
"filesystem",
"register_implementation",
"get_filesystem_class",
"get_fs_token_paths",
"get_mapper",
"open",
"open_files",
"open_local",
"registry",
"caching",
"Callback",
"available_protocols",
"available_compressions",
]
def process_entries():
if entry_points is not None:
try:
eps = entry_points()
except TypeError:
pass # importlib-metadata < 0.8
else:
if hasattr(eps, "select"): # Python 3.10+ / importlib_metadata >= 3.9.0
specs = eps.select(group="fsspec.specs")
else:
specs = eps.get("fsspec.specs", [])
registered_names = {}
for spec in specs:
err_msg = f"Unable to load filesystem from {spec}"
name = spec.name
if name in registered_names:
continue
registered_names[name] = True
register_implementation(
name,
spec.value.replace(":", "."),
errtxt=err_msg,
# We take our implementations as the ones to overload with if
# for some reason we encounter some, may be the same, already
# registered
clobber=True,
)
process_entries()
|