dvc: remove dvc.objects, fix imports

This commit is contained in:
Saugat Pachhai (सौगात)
2022-05-20 14:02:08 +05:45
committed by skshetry
parent 95c6cea36d
commit 37921d618d
103 changed files with 176 additions and 4491 deletions
+1 -1
View File
@@ -107,7 +107,7 @@ def _log_exceptions(exc: Exception) -> Optional[int]:
)
return 251
from dvc.objects.cache import DiskError
from dvc_objects.cache import DiskError
if isinstance(exc, DiskError):
from dvc.utils import relpath
+3 -3
View File
@@ -4,9 +4,9 @@ from typing import TYPE_CHECKING, Iterator, Union
from .tree import Tree
if TYPE_CHECKING:
from dvc.objects.db import ObjectDB
from dvc.objects.file import HashFile
from dvc.objects.hash_info import HashInfo
from dvc_objects.db import ObjectDB
from dvc_objects.file import HashFile
from dvc_objects.hash_info import HashInfo
logger = logging.getLogger(__name__)
+1 -1
View File
@@ -1,4 +1,4 @@
from dvc.objects._tqdm import Tqdm
from dvc_objects._tqdm import Tqdm
class QueryingProgress(Tqdm):
+3 -3
View File
@@ -2,15 +2,15 @@ import logging
from itertools import chain
from typing import TYPE_CHECKING, List, Optional
from dvc.objects.fs.callbacks import Callback
from dvc.objects.fs.generic import test_links, transfer
from dvc_objects.fs.callbacks import Callback
from dvc_objects.fs.generic import test_links, transfer
from .diff import ROOT
from .diff import diff as odiff
from .stage import stage
if TYPE_CHECKING:
from dvc.objects._ignore import Ignore
from dvc_objects._ignore import Ignore
logger = logging.getLogger(__name__)
+2 -2
View File
@@ -5,8 +5,8 @@ if TYPE_CHECKING:
def get_odb(fs, fs_path, **config):
from dvc.objects.db import ObjectDB
from dvc.objects.fs import Schemes
from dvc_objects.db import ObjectDB
from dvc_objects.fs import Schemes
from .local import LocalObjectDB
+7 -7
View File
@@ -3,10 +3,10 @@ import os
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Iterable, Set
from dvc.objects.errors import ObjectDBError
from dvc_objects.errors import ObjectDBError
if TYPE_CHECKING:
from dvc.objects.fs.base import AnyFSPath
from dvc_objects.fs.base import AnyFSPath
logger = logging.getLogger(__name__)
@@ -88,9 +88,9 @@ class ObjectDBIndex(ObjectDBIndexBase):
tmp_dir: "AnyFSPath",
name: str,
): # pylint: disable=super-init-not-called
from dvc.objects.cache import Cache, Index
from dvc.objects.fs import LocalFileSystem
from dvc.objects.fs.utils import makedirs
from dvc_objects.cache import Cache, Index
from dvc_objects.fs import LocalFileSystem
from dvc_objects.fs.utils import makedirs
self.index_dir = os.path.join(tmp_dir, self.INDEX_DIR, name)
makedirs(self.index_dir, exist_ok=True)
@@ -110,7 +110,7 @@ class ObjectDBIndex(ObjectDBIndexBase):
def clear(self):
"""Clear this index (to force re-indexing later)."""
from dvc.objects.cache import Timeout
from dvc_objects.cache import Timeout
try:
self.index.clear()
@@ -119,7 +119,7 @@ class ObjectDBIndex(ObjectDBIndexBase):
def update(self, dir_hashes: Iterable[str], file_hashes: Iterable[str]):
"""Update this index, adding the specified hashes."""
from dvc.objects.cache import Timeout
from dvc_objects.cache import Timeout
try:
with self.index.transact():
+6 -6
View File
@@ -6,11 +6,11 @@ from functools import partial
from funcy import cached_property
from shortuuid import uuid
from dvc.objects.db import ObjectDB, noop, wrap_iter
from dvc.objects.errors import ObjectDBError, ObjectFormatError
from dvc.objects.fs.system import umask
from dvc.objects.fs.utils import copyfile, relpath, remove, walk_files
from dvc.objects.hash_info import HashInfo
from dvc_objects.db import ObjectDB, noop, wrap_iter
from dvc_objects.errors import ObjectDBError, ObjectFormatError
from dvc_objects.fs.system import umask
from dvc_objects.fs.utils import copyfile, relpath, remove, walk_files
from dvc_objects.hash_info import HashInfo
logger = logging.getLogger(__name__)
@@ -49,7 +49,7 @@ class LocalObjectDB(ObjectDB):
os.chmod(to_info, self._file_mode)
def makedirs(self, fs_path):
from dvc.objects.fs.utils import makedirs
from dvc_objects.fs.utils import makedirs
makedirs(fs_path, exist_ok=True, mode=self._dir_mode)
+6 -6
View File
@@ -2,16 +2,16 @@ import logging
import os
from typing import TYPE_CHECKING, Dict
from dvc.objects.db import ObjectDB
from dvc.objects.errors import ObjectFormatError
from dvc.objects.fs import Schemes
from dvc_objects.db import ObjectDB
from dvc_objects.errors import ObjectFormatError
from dvc_objects.fs import Schemes
from ..reference import ReferenceHashFile
if TYPE_CHECKING:
from dvc.objects.fs.base import AnyFSPath, FileSystem
from dvc.objects.fs.callbacks import Callback
from dvc.objects.hash_info import HashInfo
from dvc_objects.fs.base import AnyFSPath, FileSystem
from dvc_objects.fs.callbacks import Callback
from dvc_objects.hash_info import HashInfo
logger = logging.getLogger(__name__)
+3 -3
View File
@@ -2,8 +2,8 @@ from dataclasses import dataclass, field
from typing import TYPE_CHECKING, List, Optional, Tuple
if TYPE_CHECKING:
from dvc.objects.file import HashFile
from dvc.objects.hash_info import HashInfo
from dvc_objects.file import HashFile
from dvc_objects.hash_info import HashInfo
ADD = "add"
MODIFY = "modify"
@@ -95,7 +95,7 @@ def diff(
return entry_obj.hash_info if entry_obj else None
def _in_cache(oid, cache):
from dvc.objects.errors import ObjectFormatError
from dvc_objects.errors import ObjectFormatError
if not oid:
return False
+2 -2
View File
@@ -1,5 +1,5 @@
def gc(odb, used, jobs=None, cache_odb=None, shallow=True):
from dvc.objects.errors import ObjectDBPermissionError
from dvc_objects.errors import ObjectDBPermissionError
from ._progress import QueryingProgress
from .tree import Tree
@@ -18,7 +18,7 @@ def gc(odb, used, jobs=None, cache_odb=None, shallow=True):
)
def _is_dir_hash(_hash):
from dvc.objects.hash_info import HASH_DIR_SUFFIX
from dvc_objects.hash_info import HASH_DIR_SUFFIX
return _hash.endswith(HASH_DIR_SUFFIX)
+7 -7
View File
@@ -4,14 +4,14 @@ import os
import pickle
from typing import TYPE_CHECKING, Optional
from dvc.objects.errors import ObjectFormatError
from dvc.objects.file import HashFile
from dvc.objects.fs import FS_MAP, LocalFileSystem
from dvc_objects.errors import ObjectFormatError
from dvc_objects.file import HashFile
from dvc_objects.fs import FS_MAP, LocalFileSystem
if TYPE_CHECKING:
from dvc.objects.db import ObjectDB
from dvc.objects.fs.base import AnyFSPath, FileSystem
from dvc.objects.hash_info import HashInfo
from dvc_objects.db import ObjectDB
from dvc_objects.fs.base import AnyFSPath, FileSystem
from dvc_objects.hash_info import HashInfo
logger = logging.getLogger(__name__)
@@ -77,7 +77,7 @@ class ReferenceHashFile(HashFile):
@classmethod
def from_bytes(cls, data: bytes, fs_cache: Optional[dict] = None):
from dvc.objects.fs import get_fs_cls
from dvc_objects.fs import get_fs_cls
try:
dict_ = pickle.loads(data)
+15 -15
View File
@@ -6,18 +6,18 @@ from concurrent.futures import ThreadPoolExecutor
from functools import partial
from typing import TYPE_CHECKING, Dict, Optional, Tuple
from dvc.objects._tqdm import Tqdm
from dvc.objects.file import HashFile
from dvc.objects.hash import hash_file
from dvc.objects.hash_info import HashInfo
from dvc.objects.meta import Meta
from dvc_objects._tqdm import Tqdm
from dvc_objects.file import HashFile
from dvc_objects.hash import hash_file
from dvc_objects.hash_info import HashInfo
from dvc_objects.meta import Meta
from .db.reference import ReferenceObjectDB
if TYPE_CHECKING:
from dvc.objects._ignore import Ignore
from dvc.objects.db import ObjectDB
from dvc.objects.fs.base import AnyFSPath, FileSystem
from dvc_objects._ignore import Ignore
from dvc_objects.db import ObjectDB
from dvc_objects.fs.base import AnyFSPath, FileSystem
from .tree import Tree
@@ -40,9 +40,9 @@ _STAGING_MEMFS_PATH = "dvc-staging"
def _upload_file(from_fs_path, fs, odb, upload_odb, callback=None):
from dvc.objects.fs.callbacks import Callback
from dvc.objects.fs.utils import tmp_fname
from dvc.objects.hash import HashStreamFile
from dvc_objects.fs.callbacks import Callback
from dvc_objects.fs.utils import tmp_fname
from dvc_objects.hash import HashStreamFile
fs_path = upload_odb.fs.path
tmp_info = fs_path.join(upload_odb.fs_path, tmp_fname())
@@ -186,7 +186,7 @@ _url_cache: Dict[str, str] = {}
def _make_staging_url(
fs: "FileSystem", odb: "ObjectDB", fs_path: Optional[str]
):
from dvc.objects.fs import Schemes
from dvc_objects.fs import Schemes
url = f"{Schemes.MEMORY}://{_STAGING_MEMFS_PATH}"
@@ -210,7 +210,7 @@ def _get_staging(odb: "ObjectDB") -> "ReferenceObjectDB":
Staging will be a reference ODB stored in the the global memfs.
"""
from dvc.objects.fs import MemoryFileSystem
from dvc_objects.fs import MemoryFileSystem
fs = MemoryFileSystem()
fs_path = _make_staging_url(fs, odb, odb.fs_path)
@@ -219,7 +219,7 @@ def _get_staging(odb: "ObjectDB") -> "ReferenceObjectDB":
def _load_raw_dir_obj(odb: "ObjectDB", hash_info: "HashInfo") -> "Tree":
from dvc.objects.errors import ObjectFormatError
from dvc_objects.errors import ObjectFormatError
from .tree import Tree
@@ -245,7 +245,7 @@ def _load_from_state(
name: str,
dry_run: bool,
) -> Tuple["ObjectDB", "Meta", "HashFile"]:
from dvc.objects.errors import ObjectFormatError
from dvc_objects.errors import ObjectFormatError
from . import check, load
from .tree import Tree
+4 -4
View File
@@ -1,14 +1,14 @@
import logging
from typing import TYPE_CHECKING, Dict, Iterable, NamedTuple, Optional, Set
from dvc.objects.fs import Schemes
from dvc.objects.hash_info import HashInfo
from dvc_objects.fs import Schemes
from dvc_objects.hash_info import HashInfo
from .tree import Tree
if TYPE_CHECKING:
from dvc.objects.db import ObjectDB
from dvc.objects.file import HashFile
from dvc_objects.db import ObjectDB
from dvc_objects.file import HashFile
from .db.index import ObjectDBIndexBase
+5 -5
View File
@@ -5,12 +5,12 @@ from typing import TYPE_CHECKING, Any, Callable, Iterable, Optional
from funcy import split
from dvc.objects._tqdm import Tqdm
from dvc.objects.executors import ThreadPoolExecutor
from dvc_objects._tqdm import Tqdm
from dvc_objects.executors import ThreadPoolExecutor
if TYPE_CHECKING:
from dvc.objects.db import ObjectDB
from dvc.objects.hash_info import HashInfo
from dvc_objects.db import ObjectDB
from dvc_objects.hash_info import HashInfo
from .db.index import ObjectDBIndexBase
from .status import CompareStatusResult
@@ -47,7 +47,7 @@ def _log_exceptions(func):
def find_tree_by_obj_id(
odbs: Iterable[Optional["ObjectDB"]], obj_id: "HashInfo"
) -> Optional["Tree"]:
from dvc.objects.errors import ObjectFormatError
from dvc_objects.errors import ObjectFormatError
from .tree import Tree
+9 -9
View File
@@ -5,14 +5,14 @@ from typing import TYPE_CHECKING, Dict, Iterable, Optional, Tuple
from funcy import cached_property
from dvc.objects.errors import ObjectFormatError
from dvc.objects.file import HashFile
from dvc.objects.hash import hash_file
from dvc_objects.errors import ObjectFormatError
from dvc_objects.file import HashFile
from dvc_objects.hash import hash_file
if TYPE_CHECKING:
from dvc.objects.db import ObjectDB
from dvc.objects.hash_info import HashInfo
from dvc.objects.meta import Meta
from dvc_objects.db import ObjectDB
from dvc_objects.hash_info import HashInfo
from dvc_objects.meta import Meta
logger = logging.getLogger(__name__)
@@ -59,8 +59,8 @@ class Tree(HashFile):
self._dict[key] = (meta, oid)
def digest(self, hash_info: Optional["HashInfo"] = None):
from dvc.objects.fs import MemoryFileSystem
from dvc.objects.fs.utils import tmp_fname
from dvc_objects.fs import MemoryFileSystem
from dvc_objects.fs.utils import tmp_fname
memfs = MemoryFileSystem()
fs_path = "memory://{}".format(tmp_fname(""))
@@ -131,7 +131,7 @@ class Tree(HashFile):
@classmethod
def from_list(cls, lst):
from dvc.objects.hash_info import HashInfo
from dvc_objects.hash_info import HashInfo
tree = cls(None, None, None)
for _entry in lst:
+2 -2
View File
@@ -7,8 +7,8 @@ from dvc.data.db import get_index
if TYPE_CHECKING:
from dvc.data.status import CompareStatusResult
from dvc.objects.db import ObjectDB
from dvc.objects.hash_info import HashInfo
from dvc_objects.db import ObjectDB
from dvc_objects.hash_info import HashInfo
logger = logging.getLogger(__name__)
+1 -1
View File
@@ -8,8 +8,8 @@ import dpath.util
from voluptuous import Any
from dvc.exceptions import DvcException
from dvc.objects.hash_info import HashInfo
from dvc.utils.serialize import LOADERS, ParseError
from dvc_objects.hash_info import HashInfo
from .base import Dependency
+3 -3
View File
@@ -10,9 +10,9 @@ from dvc.prompt import confirm
from .base import Dependency
if TYPE_CHECKING:
from dvc.objects.db import ObjectDB
from dvc.objects.file import HashFile
from dvc.objects.hash_info import HashInfo
from dvc_objects.db import ObjectDB
from dvc_objects.file import HashFile
from dvc_objects.hash_info import HashInfo
class RepoDependency(Dependency):
+9 -9
View File
@@ -1,8 +1,8 @@
from urllib.parse import urlparse
# pylint: disable=unused-import
from dvc.objects.fs import utils # noqa: F401
from dvc.objects.fs import ( # noqa: F401
from dvc_objects.fs import utils # noqa: F401
from dvc_objects.fs import ( # noqa: F401
FS_MAP,
AzureFileSystem,
GDriveFileSystem,
@@ -23,19 +23,19 @@ from dvc.objects.fs import ( # noqa: F401
get_fs_cls,
system,
)
from dvc.objects.fs.base import AnyFSPath, FileSystem # noqa: F401
from dvc.objects.fs.errors import ( # noqa: F401
from dvc_objects.fs.base import AnyFSPath, FileSystem # noqa: F401
from dvc_objects.fs.errors import ( # noqa: F401
AuthError,
ConfigError,
RemoteMissingDepsError,
)
from dvc.objects.fs.implementations.azure import AzureAuthError # noqa: F401
from dvc.objects.fs.implementations.gdrive import GDriveAuthError # noqa: F401
from dvc.objects.fs.implementations.local import localfs # noqa: F401
from dvc.objects.fs.implementations.ssh import ( # noqa: F401
from dvc_objects.fs.implementations.azure import AzureAuthError # noqa: F401
from dvc_objects.fs.implementations.gdrive import GDriveAuthError # noqa: F401
from dvc_objects.fs.implementations.local import localfs # noqa: F401
from dvc_objects.fs.implementations.ssh import ( # noqa: F401
DEFAULT_PORT as DEFAULT_SSH_PORT,
)
from dvc.objects.fs.path import Path # noqa: F401
from dvc_objects.fs.path import Path # noqa: F401
from .data import DataFileSystem # noqa: F401
from .dvc import DvcFileSystem # noqa: F401
+1 -1
View File
@@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Optional
from funcy import cached_property
from dvc.objects.fs.callbacks import ( # noqa: F401
from dvc_objects.fs.callbacks import ( # noqa: F401
DEFAULT_CALLBACK,
Callback,
TqdmCallback,
+3 -3
View File
@@ -5,8 +5,8 @@ import typing
from fsspec import AbstractFileSystem
from funcy import cached_property, wrap_prop
from dvc.objects.fs.base import FileSystem
from dvc.objects.fs.callbacks import DEFAULT_CALLBACK
from dvc_objects.fs.base import FileSystem
from dvc_objects.fs.callbacks import DEFAULT_CALLBACK
if typing.TYPE_CHECKING:
from dvc.types import AnyPath
@@ -117,7 +117,7 @@ class _DataFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
return bool(info.get("outs") if recurse else info.get("isout"))
def info(self, path, **kwargs):
from dvc.objects.meta import Meta
from dvc_objects.meta import Meta
key = self._get_key(path)
+6 -4
View File
@@ -10,9 +10,9 @@ from typing import TYPE_CHECKING, Callable, Optional, Tuple, Type, Union
from fsspec.spec import AbstractFileSystem
from funcy import cached_property, wrap_prop, wrap_with
from dvc.objects.fs.base import FileSystem
from dvc.objects.fs.callbacks import DEFAULT_CALLBACK
from dvc.objects.fs.path import Path
from dvc_objects.fs.base import FileSystem
from dvc_objects.fs.callbacks import DEFAULT_CALLBACK
from dvc_objects.fs.path import Path
from .data import DataFileSystem
@@ -320,7 +320,9 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
_, _, dvc_fs, dvc_path = self._get_fs_pair(path)
return dvc_fs is not None and dvc_fs.isdvc(dvc_path, **kwargs)
def ls(self, path, detail=True, dvc_only=False, **kwargs):
def ls( # pylint: disable=arguments-differ
self, path, detail=True, dvc_only=False, **kwargs
):
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
repo = dvc_fs.repo if dvc_fs else self.repo
View File
-11
View File
@@ -1,11 +0,0 @@
from typing import TYPE_CHECKING, Iterator, Protocol
if TYPE_CHECKING:
from .fs.base import AnyFSPath, FileSystem
class Ignore(Protocol):
def find(
self, fs: "FileSystem", path: "AnyFSPath"
) -> Iterator["AnyFSPath"]:
...
-161
View File
@@ -1,161 +0,0 @@
"""Manages progress bars."""
import logging
import os
import re
import sys
from threading import RLock
from tqdm import tqdm
logger = logging.getLogger(__name__)
tqdm.set_lock(RLock())
def env2bool(var, undefined=False):
"""
undefined: return value if env var is unset
"""
var = os.getenv(var, None)
if var is None:
return undefined
return bool(re.search("1|y|yes|true", var, flags=re.I))
class Tqdm(tqdm):
"""
maximum-compatibility tqdm-based progressbars
"""
BAR_FMT_DEFAULT = (
"{percentage:3.0f}% {desc}|{bar}|"
"{postfix[info]}{n_fmt}/{total_fmt}"
" [{elapsed}<{remaining}, {rate_fmt:>11}]"
)
# nested bars should have fixed bar widths to align nicely
BAR_FMT_DEFAULT_NESTED = (
"{percentage:3.0f}%|{bar:10}|{desc:{ncols_desc}.{ncols_desc}}"
"{postfix[info]}{n_fmt}/{total_fmt}"
" [{elapsed}<{remaining}, {rate_fmt:>11}]"
)
BAR_FMT_NOTOTAL = (
"{desc}{bar:b}|{postfix[info]}{n_fmt} [{elapsed}, {rate_fmt:>11}]"
)
BYTES_DEFAULTS = {
"unit": "B",
"unit_scale": True,
"unit_divisor": 1024,
"miniters": 1,
}
def __init__(
self,
iterable=None,
disable=None,
level=logging.ERROR,
desc=None,
leave=False,
bar_format=None,
bytes=False, # pylint: disable=redefined-builtin
file=None,
total=None,
postfix=None,
**kwargs,
):
"""
bytes : shortcut for
`unit='B', unit_scale=True, unit_divisor=1024, miniters=1`
desc : persists after `close()`
level : effective logging level for determining `disable`;
used only if `disable` is unspecified
disable : If (default: None) or False,
will be determined by logging level.
May be overridden to `True` due to non-TTY status.
Skip override by specifying env var `DVC_IGNORE_ISATTY`.
kwargs : anything accepted by `tqdm.tqdm()`
"""
kwargs = kwargs.copy()
if bytes:
kwargs = {**self.BYTES_DEFAULTS, **kwargs}
else:
kwargs.setdefault("unit_scale", total > 999 if total else True)
if file is None:
file = sys.stderr
# auto-disable based on `logger.level`
if not disable:
disable = logger.getEffectiveLevel() > level
# auto-disable based on TTY
if (
not disable
and not env2bool("DVC_IGNORE_ISATTY")
and hasattr(file, "isatty")
):
disable = not file.isatty()
super().__init__(
iterable=iterable,
disable=disable,
leave=leave,
desc=desc,
bar_format="!",
lock_args=(False,),
total=total,
**kwargs,
)
self.postfix = postfix or {"info": ""}
if bar_format is None:
if self.__len__():
self.bar_format = (
self.BAR_FMT_DEFAULT_NESTED
if self.pos
else self.BAR_FMT_DEFAULT
)
else:
self.bar_format = self.BAR_FMT_NOTOTAL
else:
self.bar_format = bar_format
self.refresh()
def update_to(self, current, total=None):
if total:
self.total = total
self.update(current - self.n)
def close(self):
self.postfix["info"] = ""
# remove ETA (either unknown or zero); remove completed bar
self.bar_format = self.bar_format.replace("<{remaining}", "").replace(
"|{bar:10}|", " "
)
super().close()
def wrap_fn(self, fn, callback=None):
"""
Returns a wrapped `fn` which calls `callback()` on each call.
`callback` is `self.update` by default.
"""
if callback is None:
callback = self.update
def wrapped(*args, **kwargs):
res = fn(*args, **kwargs)
callback()
return res
return wrapped
@property
def format_dict(self):
"""inject `ncols_desc` to fill the display width (`ncols`)"""
d = super().format_dict
ncols = d["ncols"] or 80
# assumes `bar_format` has max one of ("ncols_desc" & "ncols_info")
ncols_left = (
ncols - len(self.format_meter(ncols_desc=1, ncols_info=1, **d)) + 1
)
ncols_left = max(ncols_left, 0)
if ncols_left:
d["ncols_desc"] = d["ncols_info"] = ncols_left
else:
# work-around for zero-width description
d["ncols_desc"] = d["ncols_info"] = 1
d["prefix"] = ""
return d
-65
View File
@@ -1,65 +0,0 @@
import os
import pickle
from functools import wraps
from typing import Any
import diskcache
from diskcache import Disk as disk
from diskcache import Index # noqa: F401, pylint: disable=unused-import
from diskcache import Timeout # noqa: F401, pylint: disable=unused-import
# pylint: disable=redefined-builtin
class DiskError(Exception):
def __init__(self, directory: str, type: str) -> None:
self.directory = directory
self.type = type
super().__init__(f"Could not open disk '{type}' in {directory}")
def translate_pickle_error(fn):
@wraps(fn)
def wrapped(self, *args, **kwargs):
try:
return fn(self, *args, **kwargs)
except (pickle.PickleError, ValueError) as e:
if isinstance(e, ValueError) and "pickle protocol" not in str(e):
raise
# pylint: disable=protected-access
raise DiskError(self._directory, type=self._type) from e
return wrapped
class Disk(disk):
"""Reraise pickle-related errors as DiskError."""
# we need type to differentiate cache for better error messages
_type: str
put = translate_pickle_error(disk.put)
get = translate_pickle_error(disk.get)
store = translate_pickle_error(disk.store)
fetch = translate_pickle_error(disk.fetch)
class Cache(diskcache.Cache):
"""Extended to handle pickle errors and use a constant pickle protocol."""
def __init__(
self,
directory: str = None,
timeout: int = 60,
disk: disk = Disk, # pylint: disable=redefined-outer-name
type: str = None,
**settings: Any,
) -> None:
settings.setdefault("disk_pickle_protocol", 4)
super().__init__(
directory=directory, timeout=timeout, disk=disk, **settings
)
self.disk._type = self._type = type or os.path.basename(self.directory)
def __getstate__(self):
return (*super().__getstate__(), self._type)
-454
View File
@@ -1,454 +0,0 @@
import itertools
import logging
from concurrent.futures import ThreadPoolExecutor
from contextlib import suppress
from copy import copy
from functools import partial
from typing import TYPE_CHECKING, Optional
from .errors import ObjectDBPermissionError, ObjectFormatError
from .file import HashFile
if TYPE_CHECKING:
from typing import Tuple
from .fs.base import AnyFSPath, FileSystem
from .fs.callbacks import Callback
from .hash_info import HashInfo
logger = logging.getLogger(__name__)
def noop(*args, **kwargs):
pass
def wrap_iter(iterable, callback):
for index, item in enumerate(iterable, start=1):
yield item
callback(index)
class ObjectDB:
DEFAULT_VERIFY = False
DEFAULT_CACHE_TYPES = ["copy"]
CACHE_MODE: Optional[int] = None
def __init__(self, fs: "FileSystem", path: str, **config):
from .state import StateNoop
self.fs = fs
self.fs_path = path
self.state = config.get("state", StateNoop())
self.verify = config.get("verify", self.DEFAULT_VERIFY)
self.cache_types = config.get("type") or copy(self.DEFAULT_CACHE_TYPES)
self.slow_link_warning = config.get("slow_link_warning", True)
self.tmp_dir = config.get("tmp_dir")
self.read_only = config.get("read_only", False)
@property
def config(self):
return {
"state": self.state,
"verify": self.verify,
"type": self.cache_types,
"slow_link_warning": self.slow_link_warning,
"tmp_dir": self.tmp_dir,
"read_only": self.read_only,
}
def __eq__(self, other):
return (
self.fs == other.fs
and self.fs_path == other.fs_path
and self.read_only == other.read_only
)
def __hash__(self):
return hash((self.fs.protocol, self.fs_path))
def exists(self, hash_info: "HashInfo"):
return self.fs.exists(self.hash_to_path(hash_info.value))
def move(self, from_info, to_info):
self.fs.move(from_info, to_info)
def makedirs(self, fs_path):
self.fs.makedirs(fs_path)
def get(self, hash_info: "HashInfo"):
"""get raw object"""
return HashFile(
self.hash_to_path(hash_info.value),
self.fs,
hash_info,
)
def _add_file(
self,
from_fs: "FileSystem",
from_info: "AnyFSPath",
to_info: "AnyFSPath",
_hash_info: "HashInfo",
hardlink: bool = False,
callback: "Callback" = None,
):
from .fs import generic
from .fs.callbacks import Callback
self.makedirs(self.fs.path.parent(to_info))
return generic.transfer(
from_fs,
from_info,
self.fs,
to_info,
hardlink=hardlink,
callback=Callback.as_callback(callback),
)
def add(
self,
fs_path: "AnyFSPath",
fs: "FileSystem",
hash_info: "HashInfo",
hardlink: bool = False,
verify: Optional[bool] = None,
callback: "Callback" = None,
):
from .fs.callbacks import Callback
if self.read_only:
raise ObjectDBPermissionError("Cannot add to read-only ODB")
if verify is None:
verify = self.verify
try:
self.check(hash_info, check_hash=verify)
return
except (ObjectFormatError, FileNotFoundError):
pass
cache_fs_path = self.hash_to_path(hash_info.value)
with Callback.as_tqdm_callback(
callback,
desc=fs.path.name(fs_path),
bytes=True,
) as cb:
self._add_file(
fs,
fs_path,
cache_fs_path,
hash_info,
hardlink=hardlink,
callback=cb,
)
try:
if verify:
self.check(hash_info, check_hash=True)
self.protect(cache_fs_path)
self.state.save(cache_fs_path, self.fs, hash_info)
except (ObjectFormatError, FileNotFoundError):
pass
def hash_to_path(self, hash_):
return self.fs.path.join(self.fs_path, hash_[0:2], hash_[2:])
def protect(self, fs_path): # pylint: disable=unused-argument
pass
def is_protected(self, fs_path): # pylint: disable=unused-argument
return False
def unprotect(self, fs_path): # pylint: disable=unused-argument
pass
def set_exec(self, fs_path): # pylint: disable=unused-argument
pass
def check(
self,
hash_info: "HashInfo",
check_hash: bool = True,
):
"""Compare the given hash with the (corresponding) actual one if
check_hash is specified, or just verify the existence of the cache
files on the filesystem.
- Use `State` as a cache for computed hashes
+ The entries are invalidated by taking into account the following:
* mtime
* inode
* size
* hash
- Remove the file from cache if it doesn't match the actual hash
"""
obj = self.get(hash_info)
if self.is_protected(obj.fs_path):
logger.trace( # type: ignore[attr-defined]
"Assuming '%s' is unchanged since it is read-only",
obj.fs_path,
)
return
try:
obj.check(self, check_hash=check_hash)
except ObjectFormatError:
logger.debug("corrupted cache file '%s'.", obj.fs_path)
with suppress(FileNotFoundError):
self.fs.remove(obj.fs_path)
raise
if check_hash:
# making cache file read-only so we don't need to check it
# next time
self.protect(obj.fs_path)
def _list_paths(self, prefix: str = None):
prefix = prefix or ""
parts: "Tuple[str, ...]" = (self.fs_path,)
if prefix:
parts = *parts, prefix[:2]
if len(prefix) > 2:
parts = *parts, prefix[2:]
yield from self.fs.find(self.fs.path.join(*parts), prefix=bool(prefix))
def _path_to_hash(self, path):
parts = self.fs.path.parts(path)[-2:]
if not (len(parts) == 2 and parts[0] and len(parts[0]) == 2):
raise ValueError(f"Bad cache file path '{path}'")
return "".join(parts)
def _list_hashes(self, prefix=None):
"""Iterate over hashes in this fs.
If `prefix` is specified, only hashes which begin with `prefix`
will be returned.
"""
for path in self._list_paths(prefix):
try:
yield self._path_to_hash(path)
except ValueError:
logger.debug(
"'%s' doesn't look like a cache file, skipping", path
)
def _hashes_with_limit(self, limit, prefix=None):
count = 0
for hash_ in self._list_hashes(prefix):
yield hash_
count += 1
if count > limit:
logger.debug(
"`_list_hashes()` returned max '{}' hashes, "
"skipping remaining results".format(limit)
)
return
def _max_estimation_size(self, hashes):
# Max remote size allowed for us to use traverse method
return max(
self.fs.TRAVERSE_THRESHOLD_SIZE,
len(hashes)
/ self.fs.TRAVERSE_WEIGHT_MULTIPLIER
* self.fs.LIST_OBJECT_PAGE_SIZE,
)
def _estimate_remote_size(self, hashes=None, progress=noop):
"""Estimate fs size based on number of entries beginning with
"00..." prefix.
Takes a progress callback that returns current_estimated_size.
"""
prefix = "0" * self.fs.TRAVERSE_PREFIX_LEN
total_prefixes = pow(16, self.fs.TRAVERSE_PREFIX_LEN)
if hashes:
max_hashes = self._max_estimation_size(hashes)
else:
max_hashes = None
def iter_with_pbar(hashes):
total = 0
for hash_ in hashes:
total += total_prefixes
progress(total)
yield hash_
if max_hashes:
hashes = self._hashes_with_limit(
max_hashes / total_prefixes, prefix
)
else:
hashes = self._list_hashes(prefix)
remote_hashes = set(iter_with_pbar(hashes))
if remote_hashes:
remote_size = total_prefixes * len(remote_hashes)
else:
remote_size = total_prefixes
logger.debug(f"Estimated remote size: {remote_size} files")
return remote_size, remote_hashes
def _list_hashes_traverse(self, remote_size, remote_hashes, jobs=None):
"""Iterate over all hashes found in this fs.
Hashes are fetched in parallel according to prefix, except in
cases where the remote size is very small.
All hashes from the remote (including any from the size
estimation step passed via the `remote_hashes` argument) will be
returned.
NOTE: For large remotes the list of hashes will be very
big(e.g. 100M entries, md5 for each is 32 bytes, so ~3200Mb list)
and we don't really need all of it at the same time, so it makes
sense to use a generator to gradually iterate over it, without
keeping all of it in memory.
"""
num_pages = remote_size / self.fs.LIST_OBJECT_PAGE_SIZE
if num_pages < 256 / self.fs.jobs:
# Fetching prefixes in parallel requires at least 255 more
# requests, for small enough remotes it will be faster to fetch
# entire cache without splitting it into prefixes.
#
# NOTE: this ends up re-fetching hashes that were already
# fetched during remote size estimation
traverse_prefixes = [None]
else:
yield from remote_hashes
traverse_prefixes = [f"{i:02x}" for i in range(1, 256)]
if self.fs.TRAVERSE_PREFIX_LEN > 2:
traverse_prefixes += [
"{0:0{1}x}".format(i, self.fs.TRAVERSE_PREFIX_LEN)
for i in range(1, pow(16, self.fs.TRAVERSE_PREFIX_LEN - 2))
]
with ThreadPoolExecutor(max_workers=jobs or self.fs.jobs) as executor:
in_remote = executor.map(self._list_hashes, traverse_prefixes)
yield from itertools.chain.from_iterable(in_remote)
def all(self, jobs=None):
"""Iterate over all hashes in this fs.
Hashes will be fetched in parallel threads according to prefix
(except for small remotes) and a progress bar will be displayed.
"""
if not self.fs.CAN_TRAVERSE:
return self._list_hashes()
remote_size, remote_hashes = self._estimate_remote_size()
return self._list_hashes_traverse(
remote_size, remote_hashes, jobs=jobs
)
def _remove_unpacked_dir(self, hash_):
pass
def list_hashes_exists(self, hashes, jobs=None):
"""Return list of the specified hashes which exist in this fs.
Hashes will be queried individually.
"""
logger.debug(f"Querying {len(hashes)} hashes via object_exists")
with ThreadPoolExecutor(max_workers=jobs or self.fs.jobs) as executor:
fs_paths = map(self.hash_to_path, hashes)
in_remote = executor.map(self.fs.exists, fs_paths)
yield from itertools.compress(hashes, in_remote)
def hashes_exist(self, hashes, jobs=None, progress=noop):
"""Check if the given hashes are stored in the remote.
There are two ways of performing this check:
- Traverse method: Get a list of all the files in the remote
(traversing the cache directory) and compare it with
the given hashes. Cache entries will be retrieved in parallel
threads according to prefix (i.e. entries starting with, "00...",
"01...", and so on) and a progress bar will be displayed.
- Exists method: For each given hash, run the `exists`
method and filter the hashes that aren't on the remote.
This is done in parallel threads.
It also shows a progress bar when performing the check.
The reason for such an odd logic is that most of the remotes
take much shorter time to just retrieve everything they have under
a certain prefix (e.g. s3, gs, ssh, hdfs). Other remotes that can
check if particular file exists much quicker, use their own
implementation of hashes_exist (see ssh, local).
Which method to use will be automatically determined after estimating
the size of the remote cache, and comparing the estimated size with
len(hashes). To estimate the size of the remote cache, we fetch
a small subset of cache entries (i.e. entries starting with "00...").
Based on the number of entries in that subset, the size of the full
cache can be estimated, since the cache is evenly distributed according
to hash.
Takes a callback that returns value in the format of:
(phase, total, current). The phase can be {"estimating, "querying"}.
Returns:
A list with hashes that were found in the remote
"""
# Remotes which do not use traverse prefix should override
# hashes_exist() (see ssh, local)
assert self.fs.TRAVERSE_PREFIX_LEN >= 2
# During the tests, for ensuring that the traverse behavior
# is working we turn on this option. It will ensure the
# _list_hashes_traverse() is called.
always_traverse = getattr(self.fs, "_ALWAYS_TRAVERSE", False)
hashes = set(hashes)
if (
len(hashes) == 1 or not self.fs.CAN_TRAVERSE
) and not always_traverse:
remote_hashes = self.list_hashes_exists(hashes, jobs)
callback = partial(progress, "querying", len(hashes))
return list(wrap_iter(remote_hashes, callback))
# Max remote size allowed for us to use traverse method
estimator_cb = partial(progress, "estimating", None)
remote_size, remote_hashes = self._estimate_remote_size(
hashes, progress=estimator_cb
)
traverse_pages = remote_size / self.fs.LIST_OBJECT_PAGE_SIZE
# For sufficiently large remotes, traverse must be weighted to account
# for performance overhead from large lists/sets.
# From testing with S3, for remotes with 1M+ files, object_exists is
# faster until len(hashes) is at least 10k~100k
if remote_size > self.fs.TRAVERSE_THRESHOLD_SIZE:
traverse_weight = (
traverse_pages * self.fs.TRAVERSE_WEIGHT_MULTIPLIER
)
else:
traverse_weight = traverse_pages
if len(hashes) < traverse_weight and not always_traverse:
logger.debug(
"Large remote ('{}' hashes < '{}' traverse weight), "
"using object_exists for remaining hashes".format(
len(hashes), traverse_weight
)
)
remaining_hashes = hashes - remote_hashes
ret = list(hashes & remote_hashes)
callback = partial(progress, "querying", len(remaining_hashes))
ret.extend(
wrap_iter(
self.list_hashes_exists(remaining_hashes, jobs), callback
)
)
return ret
logger.debug(f"Querying '{len(hashes)}' hashes via traverse")
remote_hashes = self._list_hashes_traverse(
remote_size, remote_hashes, jobs=jobs
)
callback = partial(progress, "querying", remote_size)
return list(hashes & set(wrap_iter(remote_hashes, callback)))
-10
View File
@@ -1,10 +0,0 @@
class ObjectFormatError(Exception):
pass
class ObjectDBError(Exception):
pass
class ObjectDBPermissionError(ObjectDBError):
pass
-74
View File
@@ -1,74 +0,0 @@
import queue
import sys
from concurrent import futures
from itertools import islice
from typing import Any, Callable, Iterable, Iterator, Set, TypeVar
_T = TypeVar("_T")
class ThreadPoolExecutor(futures.ThreadPoolExecutor):
_max_workers: int
def __init__(
self, max_workers: int = None, cancel_on_error: bool = False, **kwargs
):
super().__init__(max_workers=max_workers, **kwargs)
self._cancel_on_error = cancel_on_error
@property
def max_workers(self) -> int:
return self._max_workers
def imap_unordered(
self, fn: Callable[..., _T], *iterables: Iterable[Any]
) -> Iterator[_T]:
"""Lazier version of map that does not preserve ordering of results.
It does not create all the futures at once to reduce memory usage.
"""
def create_taskset(n: int) -> Set[futures.Future]:
return {self.submit(fn, *args) for args in islice(it, n)}
it = zip(*iterables)
tasks = create_taskset(self.max_workers * 5)
while tasks:
done, tasks = futures.wait(
tasks, return_when=futures.FIRST_COMPLETED
)
for fut in done:
yield fut.result()
tasks.update(create_taskset(len(done)))
def shutdown(self, wait=True, *, cancel_futures=False):
if sys.version_info > (3, 9):
# pylint: disable=unexpected-keyword-arg
return super().shutdown(wait=wait, cancel_futures=cancel_futures)
with self._shutdown_lock:
self._shutdown = True
if cancel_futures:
# Drain all work items from the queue, and then cancel their
# associated futures.
while True:
try:
work_item = self._work_queue.get_nowait()
except queue.Empty:
break
if work_item is not None:
work_item.future.cancel()
# Send a wake-up to prevent threads calling
# _work_queue.get(block=True) from permanently blocking.
self._work_queue.put(None)
if wait:
for t in self._threads:
t.join()
def __exit__(self, exc_type, exc_val, exc_tb):
if self._cancel_on_error:
self.shutdown(wait=True, cancel_futures=exc_val is not None)
else:
self.shutdown(wait=True)
return False
-83
View File
@@ -1,83 +0,0 @@
import errno
import logging
import os
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from .db import ObjectDB
from .fs.base import AnyFSPath, FileSystem
from .hash_info import HashInfo
logger = logging.getLogger(__name__)
class HashFile:
def __init__(
self,
fs_path: "AnyFSPath",
fs: Optional["FileSystem"],
hash_info: "HashInfo",
name: Optional[str] = None,
):
self.fs_path = fs_path
self.fs = fs
self.hash_info = hash_info
self.name = name
def __len__(self):
return 1
def __str__(self):
return f"object {self.hash_info}"
def __bool__(self):
return bool(self.hash_info)
def __eq__(self, other):
if not isinstance(other, HashFile):
return False
return (
self.fs_path == other.fs_path
and self.fs == other.fs
and self.hash_info == other.hash_info
)
def __hash__(self):
return hash(
(
self.hash_info,
self.fs_path,
self.fs.protocol if self.fs else None,
)
)
def check(self, odb: "ObjectDB", check_hash: bool = True):
if not check_hash:
assert self.fs
if not self.fs.exists(self.fs_path):
raise FileNotFoundError(
errno.ENOENT, os.strerror(errno.ENOENT), self.fs_path
)
else:
return None
self._check_hash(odb)
def _check_hash(self, odb):
from .errors import ObjectFormatError
from .hash import hash_file
_, actual = hash_file(
self.fs_path, self.fs, self.hash_info.name, odb.state
)
logger.trace(
"cache '%s' expected '%s' actual '%s'",
self.fs_path,
self.hash_info,
actual,
)
assert actual.name == self.hash_info.name
if actual.value.split(".")[0] != self.hash_info.value.split(".")[0]:
raise ObjectFormatError(f"{self} is corrupted")
-54
View File
@@ -1,54 +0,0 @@
from urllib.parse import urlparse
from . import generic # noqa: F401
from .implementations.azure import AzureFileSystem
from .implementations.gdrive import GDriveFileSystem
from .implementations.gs import GSFileSystem
from .implementations.hdfs import HDFSFileSystem
from .implementations.http import HTTPFileSystem
from .implementations.https import HTTPSFileSystem
from .implementations.local import LocalFileSystem
from .implementations.memory import MemoryFileSystem # noqa: F401
from .implementations.oss import OSSFileSystem
from .implementations.s3 import S3FileSystem
from .implementations.ssh import SSHFileSystem
from .implementations.webdav import WebDAVFileSystem, WebDAVSFileSystem
from .implementations.webhdfs import WebHDFSFileSystem
from .scheme import Schemes
FS_MAP = {
Schemes.AZURE: AzureFileSystem,
Schemes.GDRIVE: GDriveFileSystem,
Schemes.GS: GSFileSystem,
Schemes.HDFS: HDFSFileSystem,
Schemes.WEBHDFS: WebHDFSFileSystem,
Schemes.HTTP: HTTPFileSystem,
Schemes.HTTPS: HTTPSFileSystem,
Schemes.S3: S3FileSystem,
Schemes.SSH: SSHFileSystem,
Schemes.OSS: OSSFileSystem,
Schemes.WEBDAV: WebDAVFileSystem,
Schemes.WEBDAVS: WebDAVSFileSystem,
# NOTE: LocalFileSystem is the default
}
def _import_class(cls: str):
"""Take a string FQP and return the imported class or identifier
clas is of the form "package.module.klass".
"""
import importlib
mod, name = cls.rsplit(".", maxsplit=1)
module = importlib.import_module(mod)
return getattr(module, name)
def get_fs_cls(remote_conf, cls=None, scheme=None):
if cls:
return _import_class(cls)
if not scheme:
scheme = urlparse(remote_conf["url"]).scheme
return FS_MAP.get(scheme, LocalFileSystem)
-569
View File
@@ -1,569 +0,0 @@
import datetime
import logging
import os
import shutil
from multiprocessing import cpu_count
from typing import (
IO,
TYPE_CHECKING,
Any,
ClassVar,
Dict,
Iterator,
List,
Optional,
Union,
cast,
overload,
)
from funcy import cached_property
from ..executors import ThreadPoolExecutor
from .callbacks import DEFAULT_CALLBACK, Callback
from .errors import RemoteMissingDepsError
if TYPE_CHECKING:
from typing import BinaryIO, TextIO
from fsspec.spec import AbstractFileSystem
from typing_extensions import Literal
logger = logging.getLogger(__name__)
FSPath = str
AnyFSPath = str
# An info() entry, might evolve to a TypedDict
# in the future (e.g for properly type 'size' etc).
Entry = Dict[str, Any]
class LinkError(OSError):
def __init__(self, link: str, fs: "FileSystem", path: str) -> None:
import errno
super().__init__(
errno.EPERM,
f"{link} is not supported for {fs.protocol} by {type(fs)}",
path,
)
class FileSystem:
sep = "/"
protocol = "base"
REQUIRES: ClassVar[Dict[str, str]] = {}
_JOBS = 4 * cpu_count()
HASH_JOBS = max(1, min(4, cpu_count() // 2))
LIST_OBJECT_PAGE_SIZE = 1000
TRAVERSE_WEIGHT_MULTIPLIER = 5
TRAVERSE_PREFIX_LEN = 2
TRAVERSE_THRESHOLD_SIZE = 500000
CAN_TRAVERSE = True
PARAM_CHECKSUM: ClassVar[Optional[str]] = None
def __init__(self, **kwargs):
self._check_requires(**kwargs)
self.jobs = kwargs.get("jobs") or self._JOBS
self.hash_jobs = kwargs.get("checksum_jobs") or self.HASH_JOBS
self._config = kwargs
self.fs_args = {"skip_instance_cache": True}
self.fs_args.update(self._prepare_credentials(**kwargs))
@property
def config(self) -> Dict[str, Any]:
return self._config
@cached_property
def path(self):
from .path import Path
def _getcwd():
return self.fs.root_marker
return Path(self.sep, getcwd=_getcwd)
@classmethod
def _strip_protocol(cls, path: str) -> str:
return path
def unstrip_protocol(self, path: str) -> str:
return path
@cached_property
def fs(self) -> "AbstractFileSystem":
raise NotImplementedError
@staticmethod
def _get_kwargs_from_urls(urlpath: str) -> "Dict[str, Any]":
from fsspec.utils import infer_storage_options
options = infer_storage_options(urlpath)
options.pop("path", None)
options.pop("protocol", None)
return options
def _prepare_credentials(
self, **config: Dict[str, Any] # pylint: disable=unused-argument
) -> Dict[str, Any]:
"""Prepare the arguments for authentication to the
host filesystem"""
return {}
@classmethod
def get_missing_deps(cls) -> List[str]:
import importlib
missing: List[str] = []
for package, module in cls.REQUIRES.items():
try:
importlib.import_module(module)
except ImportError:
missing.append(package)
return missing
def _check_requires(self, **kwargs):
from .scheme import Schemes
missing = self.get_missing_deps()
if not missing:
return
proto = self.protocol
if proto == Schemes.WEBDAVS:
proto = Schemes.WEBDAV
url = kwargs.get("url", f"{self.protocol}://")
raise RemoteMissingDepsError(self, proto, url, missing)
def isdir(self, path: AnyFSPath) -> bool:
return self.fs.isdir(path)
def isfile(self, path: AnyFSPath) -> bool:
return self.fs.isfile(path)
def is_empty(self, path: AnyFSPath) -> bool:
entry = self.info(path)
if entry["type"] == "directory":
return not self.fs.ls(path)
return entry["size"] == 0
@overload
def open(
self,
path: AnyFSPath,
mode: "Literal['rb', 'br', 'wb']",
**kwargs: Any,
) -> "BinaryIO": # pylint: disable=arguments-differ
return self.open(path, mode, **kwargs)
@overload
def open(
self,
path: AnyFSPath,
mode: "Literal['r', 'rt', 'w']",
**kwargs: Any,
) -> "TextIO": # pylint: disable=arguments-differ
...
def open(
self,
path: AnyFSPath,
mode: str = "r",
**kwargs: Any,
) -> "IO[Any]": # pylint: disable=arguments-differ
if "b" in mode:
kwargs.pop("encoding", None)
return self.fs.open(path, mode=mode, **kwargs)
def read_block(
self,
path: AnyFSPath,
offset: int,
length: int,
delimiter: bytes = None,
) -> bytes:
return self.fs.read_block(path, offset, length, delimiter=delimiter)
def cat(
self,
path: Union[AnyFSPath, List[AnyFSPath]],
recursive: bool = False,
on_error: "Literal['raise', 'omit', 'return']" = "raise",
**kwargs: Any,
) -> Union[bytes, Dict[AnyFSPath, bytes]]:
return self.fs.cat(
path, recursive=recursive, on_error=on_error, **kwargs
)
def cat_ranges(
self,
paths: List[AnyFSPath],
starts: List[int],
ends: List[int],
max_gap: int = None,
**kwargs,
) -> List[bytes]:
return self.fs.cat_ranges(
paths, starts, ends, max_gap=max_gap, **kwargs
)
def cat_file(
self,
path: AnyFSPath,
start: int = None,
end: int = None,
**kwargs: Any,
) -> bytes:
return self.fs.cat_file(path, start=start, end=end, **kwargs)
def head(self, path: AnyFSPath, size: int = 1024) -> bytes:
return self.fs.head(path, size=size)
def tail(self, path: AnyFSPath, size: int = 1024) -> bytes:
return self.fs.tail(path, size=size)
def pipe_file(self, path: AnyFSPath, value: bytes, **kwargs: Any) -> None:
return self.fs.pipe_file(path, value, **kwargs)
def pipe(
self,
path: Union[AnyFSPath, Dict[AnyFSPath, bytes]],
value: Optional[bytes] = None,
**kwargs: Any,
) -> None:
return self.fs.pipe(path, value=value, **kwargs)
def touch(
self, path: AnyFSPath, truncate: bool = True, **kwargs: Any
) -> None:
return self.fs.touch(path, truncate=truncate, **kwargs)
def checksum(self, path: AnyFSPath) -> str:
return self.fs.checksum(path)
def copy(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None:
self.makedirs(self.path.parent(to_info))
self.fs.copy(from_info, to_info)
def cp_file(
self, from_info: AnyFSPath, to_info: AnyFSPath, **kwargs: Any
) -> None:
self.fs.cp_file(from_info, to_info, **kwargs)
def exists(self, path: AnyFSPath) -> bool:
return self.fs.exists(path)
def lexists(self, path: AnyFSPath) -> bool:
return self.fs.lexists(path)
def symlink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None:
try:
return self.fs.symlink(from_info, to_info)
except AttributeError:
raise LinkError("symlink", self, from_info)
def hardlink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None:
try:
return self.fs.hardlink(from_info, to_info)
except AttributeError:
raise LinkError("symlink", self, from_info)
def reflink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None:
try:
return self.fs.reflink(from_info, to_info)
except AttributeError:
raise LinkError("symlink", self, from_info)
def is_symlink(self, path: AnyFSPath) -> bool:
try:
return self.fs.is_symlink(path)
except AttributeError:
return False
def is_hardlink(self, path: AnyFSPath) -> bool:
try:
return self.fs.is_hardlink(path)
except AttributeError:
return False
def iscopy(self, path: AnyFSPath) -> bool:
return not (self.is_symlink(path) or self.is_hardlink(path))
@overload
def ls(
self, path: AnyFSPath, detail: "Literal[True]"
) -> "Iterator[Entry]":
...
@overload
def ls(self, path: AnyFSPath, detail: "Literal[False]") -> Iterator[str]:
...
def ls(self, path, detail=False, **kwargs):
return self.fs.ls(path, detail=detail)
def find(
self,
path: AnyFSPath,
prefix: bool = False, # pylint: disable=unused-argument
) -> Iterator[str]:
yield from self.fs.find(path)
def mv(
self, from_info: AnyFSPath, to_info: AnyFSPath, **kwargs: Any
) -> None:
self.fs.mv(from_info, to_info)
move = mv
def rmdir(self, path: AnyFSPath) -> None:
self.fs.rmdir(path)
def rm_file(self, path: AnyFSPath) -> None:
self.fs.rm_file(path)
def rm(self, path: AnyFSPath) -> None:
self.fs.rm(path, recursive=True)
remove = rm
def info(self, path: AnyFSPath) -> "Entry":
return self.fs.info(path)
def mkdir(
self, path: AnyFSPath, create_parents: bool = True, **kwargs: Any
) -> None:
self.fs.mkdir(path, create_parents=create_parents, **kwargs)
def makedirs(self, path: AnyFSPath, **kwargs: Any) -> None:
self.fs.makedirs(path, exist_ok=kwargs.pop("exist_ok", True))
def put_file(
self,
from_file: Union[AnyFSPath, "BinaryIO"],
to_info: AnyFSPath,
callback: Callback = DEFAULT_CALLBACK,
size: int = None,
**kwargs,
) -> None:
if size:
callback.set_size(size)
if hasattr(from_file, "read"):
stream = callback.wrap_attr(cast("BinaryIO", from_file))
self.upload_fobj(stream, to_info, size=size)
else:
assert isinstance(from_file, str)
self.fs.put_file(
os.fspath(from_file), to_info, callback=callback, **kwargs
)
self.fs.invalidate_cache(self.path.parent(to_info))
def get_file(
self,
from_info: AnyFSPath,
to_info: AnyFSPath,
callback: Callback = DEFAULT_CALLBACK,
**kwargs,
) -> None:
self.fs.get_file(from_info, to_info, callback=callback, **kwargs)
def upload_fobj(self, fobj: IO, to_info: AnyFSPath, **kwargs) -> None:
self.makedirs(self.path.parent(to_info))
with self.open(to_info, "wb") as fdest:
shutil.copyfileobj(
fobj,
fdest,
length=getattr(fdest, "blocksize", None), # type: ignore
)
def walk(
self,
path: AnyFSPath,
topdown: bool = True,
**kwargs: Any,
):
return self.fs.walk(path, topdown=topdown, **kwargs)
def glob(self, path: AnyFSPath, **kwargs: Any):
return self.fs.glob(path, **kwargs)
def size(self, path: AnyFSPath) -> Optional[int]:
return self.fs.size(path)
def sizes(self, paths: List[AnyFSPath]) -> List[Optional[int]]:
return self.fs.sizes(paths)
def du(
self,
path: AnyFSPath,
total: bool = True,
maxdepth: int = None,
**kwargs: Any,
) -> Union[int, Dict[AnyFSPath, int]]:
return self.fs.du(path, total=total, maxdepth=maxdepth, **kwargs)
def put(
self,
from_info: Union[AnyFSPath, List[AnyFSPath]],
to_info: Union[AnyFSPath, List[AnyFSPath]],
callback: "Callback" = DEFAULT_CALLBACK,
recursive: bool = False, # pylint: disable=unused-argument
batch_size: int = None,
):
jobs = batch_size or self.jobs
if self.fs.async_impl:
return self.fs.put(
from_info,
to_info,
callback=callback,
batch_size=jobs,
recursive=recursive,
)
assert not recursive, "not implemented yet"
from_infos = [from_info] if isinstance(from_info, str) else from_info
to_infos = [to_info] if isinstance(to_info, str) else to_info
callback.set_size(len(from_infos))
executor = ThreadPoolExecutor(max_workers=jobs, cancel_on_error=True)
with executor:
put_file = callback.wrap_and_branch(self.put_file)
list(executor.imap_unordered(put_file, from_infos, to_infos))
def get(
self,
from_info: Union[AnyFSPath, List[AnyFSPath]],
to_info: Union[AnyFSPath, List[AnyFSPath]],
callback: "Callback" = DEFAULT_CALLBACK,
recursive: bool = False, # pylint: disable=unused-argument
batch_size: int = None,
) -> None:
# Currently, the implementation is non-recursive if the paths are
# provided as a list, and recursive if it's a single path.
from .implementations.local import localfs
def get_file(rpath, lpath, **kwargs):
localfs.makedirs(localfs.path.parent(lpath), exist_ok=True)
self.fs.get_file(rpath, lpath, **kwargs)
get_file = callback.wrap_and_branch(get_file)
if isinstance(from_info, list) and isinstance(to_info, list):
from_infos: List[AnyFSPath] = from_info
to_infos: List[AnyFSPath] = to_info
else:
assert isinstance(from_info, str)
assert isinstance(to_info, str)
if not self.isdir(from_info):
callback.set_size(1)
return get_file(from_info, to_info)
from_infos = list(self.find(from_info))
if not from_infos:
return localfs.makedirs(to_info, exist_ok=True)
to_infos = [
localfs.path.join(
to_info, *self.path.relparts(info, from_info)
)
for info in from_infos
]
jobs = batch_size or self.jobs
if self.fs.async_impl:
return self.fs.get(
from_infos,
to_infos,
callback=callback,
batch_size=jobs,
)
callback.set_size(len(from_infos))
executor = ThreadPoolExecutor(max_workers=jobs, cancel_on_error=True)
with executor:
list(executor.imap_unordered(get_file, from_infos, to_infos))
def ukey(self, path: AnyFSPath) -> str:
return self.fs.ukey(path)
def created(self, path: AnyFSPath) -> datetime.datetime:
return self.fs.created(path)
def modified(self, path: AnyFSPath) -> datetime.datetime:
return self.fs.modified(path)
def sign(
self, path: AnyFSPath, expiration: int = 100, **kwargs: Any
) -> str:
return self.fs.sign(path, expiration=expiration, **kwargs)
class ObjectFileSystem(FileSystem): # pylint: disable=abstract-method
TRAVERSE_PREFIX_LEN = 3
def makedirs(self, path: AnyFSPath, **kwargs: Any) -> None:
# For object storages make this method a no-op. The original
# fs.makedirs() method will only check if the bucket exists
# and create if it doesn't though we don't want to support
# that behavior, and the check will cost some time so we'll
# simply ignore all mkdir()/makedirs() calls.
return None
def mkdir(
self, path: AnyFSPath, create_parents: bool = True, **kwargs: Any
) -> None:
return None
def _isdir(self, path: AnyFSPath) -> bool:
# Directory in object storages are interpreted differently
# among different fsspec providers, so this logic is a temporary
# measure for us to adapt as of now. It checks whether it is a
# directory (as in a prefix with contents) or whether it is an empty
# file where it's name ends with a forward slash
entry = self.info(path)
return entry["type"] == "directory" or (
entry["size"] == 0
and entry["type"] == "file"
and entry["name"].endswith("/")
)
def isdir(self, path: AnyFSPath) -> bool:
try:
return self._isdir(path)
except FileNotFoundError:
return False
def isfile(self, path: AnyFSPath) -> bool:
try:
return not self._isdir(path)
except FileNotFoundError:
return False
def find(self, path: AnyFSPath, prefix: bool = False) -> Iterator[str]:
if prefix:
with_prefix = self.path.parent(path)
files = self.fs.find(with_prefix, prefix=self.path.parts(path)[-1])
else:
with_prefix = path
files = self.fs.find(path)
# When calling find() on a file, it returns the same file in a list.
# For object-based storages, the same behavior applies to empty
# directories since they are represented as files. This condition
# checks whether we should yield an empty list (if it is an empty
# directory) or just yield the file itself.
if len(files) == 1 and files[0] == with_prefix and self.isdir(path):
return None
yield from files
-161
View File
@@ -1,161 +0,0 @@
from contextlib import ExitStack
from functools import wraps
from typing import TYPE_CHECKING, Any, Dict, Optional, TypeVar, overload
import fsspec
from funcy import cached_property
if TYPE_CHECKING:
from typing import BinaryIO, Callable, TextIO, Union
from typing_extensions import ParamSpec
from .._tqdm import Tqdm
_P = ParamSpec("_P")
_R = TypeVar("_R")
class Callback(fsspec.Callback):
"""Callback usable as a context manager, and a few helper methods."""
@overload
def wrap_attr(self, fobj: "BinaryIO", method: str = "read") -> "BinaryIO":
...
@overload
def wrap_attr(self, fobj: "TextIO", method: str = "read") -> "TextIO":
...
def wrap_attr(
self, fobj: "Union[TextIO, BinaryIO]", method: str = "read"
) -> "Union[TextIO, BinaryIO]":
from tqdm.utils import CallbackIOWrapper
wrapped = CallbackIOWrapper(self.relative_update, fobj, method)
return wrapped
def wrap_fn(self, fn: "Callable[_P, _R]") -> "Callable[_P, _R]":
@wraps(fn)
def wrapped(*args: "_P.args", **kwargs: "_P.kwargs") -> "_R":
res = fn(*args, **kwargs)
self.relative_update()
return res
return wrapped
def wrap_and_branch(self, fn: "Callable") -> "Callable":
"""
Wraps a function, and pass a new child callback to it.
When the function completes, we increment the parent callback by 1.
"""
wrapped = self.wrap_fn(fn)
@wraps(fn)
def func(path1: str, path2: str):
kw: Dict[str, Any] = {}
with self.branch(path1, path2, kw):
return wrapped(path1, path2, **kw)
return func
def __enter__(self):
return self
def __exit__(self, *exc_args):
self.close()
def close(self):
"""Handle here on exit."""
def relative_update(self, inc: int = 1) -> None:
inc = inc if inc is not None else 0
return super().relative_update(inc)
def absolute_update(self, value: int) -> None:
value = value if value is not None else self.value
return super().absolute_update(value)
@classmethod
def as_callback(
cls, maybe_callback: Optional["Callback"] = None
) -> "Callback":
if maybe_callback is None:
return DEFAULT_CALLBACK
return maybe_callback
@classmethod
def as_tqdm_callback(
cls,
callback: Optional["Callback"] = None,
**tqdm_kwargs: Any,
) -> "Callback":
return callback or TqdmCallback(**tqdm_kwargs)
def branch(
self,
path_1: str,
path_2: str,
kwargs: Dict[str, Any],
child: "Callback" = None,
) -> "Callback":
child = kwargs["callback"] = child or DEFAULT_CALLBACK
return child
class NoOpCallback(Callback, fsspec.callbacks.NoOpCallback):
pass
class TqdmCallback(Callback):
def __init__(
self,
size: Optional[int] = None,
value: int = 0,
progress_bar: "Tqdm" = None,
**tqdm_kwargs,
):
tqdm_kwargs["total"] = size or -1
self._tqdm_kwargs = tqdm_kwargs
self._progress_bar = progress_bar
self._stack = ExitStack()
super().__init__(size=size, value=value)
@cached_property
def progress_bar(self):
from .._tqdm import Tqdm
progress_bar = (
self._progress_bar
if self._progress_bar is not None
else Tqdm(**self._tqdm_kwargs)
)
return self._stack.enter_context(progress_bar)
def __enter__(self):
return self
def close(self):
self._stack.close()
def set_size(self, size):
# Tqdm tries to be smart when to refresh,
# so we try to force it to re-render.
super().set_size(size)
self.progress_bar.refresh()
def call(self, hook_name=None, **kwargs):
self.progress_bar.update_to(self.value, total=self.size)
def branch(
self,
path_1: str,
path_2: str,
kwargs,
child: Optional[Callback] = None,
):
child = child or TqdmCallback(bytes=True, desc=path_1)
return super().branch(path_1, path_2, kwargs, child=child)
DEFAULT_CALLBACK = NoOpCallback()
-29
View File
@@ -1,29 +0,0 @@
from typing import TYPE_CHECKING, List
if TYPE_CHECKING:
from .base import FileSystem
class AuthError(Exception):
pass
class ConfigError(Exception):
pass
class RemoteMissingDepsError(Exception):
def __init__(
self,
fs: "FileSystem",
protocol: str,
url: str,
missing: List[str] = None,
) -> None:
self.protocol = protocol
self.fs = fs
self.url = url
self.missing_deps = missing or []
super().__init__(
f"filesystem for '{protocol}': '{type(fs)}' missing dependencies"
)
-172
View File
@@ -1,172 +0,0 @@
import errno
import logging
import os
from typing import TYPE_CHECKING, List, Optional
from .callbacks import DEFAULT_CALLBACK
from .implementations.local import LocalFileSystem
from .utils import as_atomic
if TYPE_CHECKING:
from .base import AnyFSPath, FileSystem
from .callbacks import Callback
logger = logging.getLogger(__name__)
def _link(
link: "str",
from_fs: "FileSystem",
from_path: "AnyFSPath",
to_fs: "FileSystem",
to_path: "AnyFSPath",
) -> None:
if not isinstance(from_fs, type(to_fs)):
raise OSError(errno.EXDEV, "can't link across filesystems")
func = getattr(to_fs, link)
func(from_path, to_path)
def copy(
from_fs: "FileSystem",
from_path: "AnyFSPath",
to_fs: "FileSystem",
to_path: "AnyFSPath",
callback: "Callback" = DEFAULT_CALLBACK,
) -> None:
if isinstance(from_fs, LocalFileSystem):
return to_fs.put_file(from_path, to_path, callback=callback)
if isinstance(to_fs, LocalFileSystem):
from .implementations.local import localfs
with as_atomic(localfs, to_path, create_parents=True) as tmp_file:
return from_fs.get_file(from_path, tmp_file, callback=callback)
with from_fs.open(from_path, mode="rb") as fobj:
size = from_fs.size(from_path)
return to_fs.put_file(fobj, to_path, size=size, callback=callback)
def _try_links(
links: List["str"],
from_fs: "FileSystem",
from_path: "AnyFSPath",
to_fs: "FileSystem",
to_path: "AnyFSPath",
callback: "Callback" = DEFAULT_CALLBACK,
) -> None:
error = None
while links:
link = links[0]
if link == "copy":
return copy(from_fs, from_path, to_fs, to_path, callback=callback)
try:
return _link(link, from_fs, from_path, to_fs, to_path)
except OSError as exc:
if exc.errno not in [errno.EPERM, errno.ENOTSUP, errno.EXDEV]:
raise
error = exc
del links[0]
raise OSError(
errno.ENOTSUP, "no more link types left to try out"
) from error
def transfer(
from_fs: "FileSystem",
from_path: "AnyFSPath",
to_fs: "FileSystem",
to_path: "AnyFSPath",
hardlink: bool = False,
links: Optional[List["str"]] = None,
callback: "Callback" = DEFAULT_CALLBACK,
) -> None:
try:
assert not (hardlink and links)
if hardlink:
links = links or ["reflink", "hardlink", "copy"]
else:
links = links or ["reflink", "copy"]
_try_links(
links, from_fs, from_path, to_fs, to_path, callback=callback
)
except OSError as exc:
# If the target file already exists, we are going to simply
# ignore the exception (#4992).
#
# On Windows, it is not always guaranteed that you'll get
# FileExistsError (it might be PermissionError or a bare OSError)
# but all of those exceptions raised from the original
# FileExistsError so we have a separate check for that.
if isinstance(exc, FileExistsError) or (
os.name == "nt"
and exc.__context__
and isinstance(exc.__context__, FileExistsError)
):
logger.debug("'%s' file already exists, skipping", to_path)
return None
raise
def _test_link(
link: "str",
from_fs: "FileSystem",
from_file: "AnyFSPath",
to_fs: "FileSystem",
to_file: "AnyFSPath",
) -> bool:
try:
_try_links([link], from_fs, from_file, to_fs, to_file)
except OSError:
logger.debug("", exc_info=True)
return False
try:
_is_link_func = getattr(to_fs, f"is_{link}")
return _is_link_func(to_file)
except AttributeError:
pass
return True
def test_links(
links: List["str"],
from_fs: "FileSystem",
from_path: "AnyFSPath",
to_fs: "FileSystem",
to_path: "AnyFSPath",
) -> List["AnyFSPath"]:
from .utils import tmp_fname
from_file = from_fs.path.join(from_path, tmp_fname())
to_file = to_fs.path.join(
to_fs.path.parent(to_path),
tmp_fname(),
)
from_fs.makedirs(from_fs.path.parent(from_file))
with from_fs.open(from_file, "wb") as fobj:
fobj.write(b"test")
to_fs.makedirs(to_fs.path.parent(to_file))
ret = []
try:
for link in links:
try:
if _test_link(link, from_fs, from_file, to_fs, to_file):
ret.append(link)
finally:
to_fs.remove(to_file)
finally:
from_fs.remove(from_file)
return ret
-154
View File
@@ -1,154 +0,0 @@
import logging
import os
import threading
from fsspec.asyn import fsspec_loop
from fsspec.utils import infer_storage_options
from funcy import cached_property, memoize, wrap_prop
from ..base import ObjectFileSystem
from ..errors import AuthError
logger = logging.getLogger(__name__)
_DEFAULT_CREDS_STEPS = (
"https://azuresdkdocs.blob.core.windows.net/$web/python/"
"azure-identity/1.4.0/azure.identity.html#azure.identity"
".DefaultAzureCredential"
)
class AzureAuthError(AuthError):
pass
@memoize
def _az_config():
# NOTE: ideally we would've used get_default_cli().config from
# azure.cli.core, but azure-cli-core has a lot of conflicts with other
# dependencies. So instead we are just use knack directly
from knack.config import CLIConfig
config_dir = os.getenv(
"AZURE_CONFIG_DIR", os.path.expanduser(os.path.join("~", ".azure"))
)
return CLIConfig(config_dir=config_dir, config_env_var_prefix="AZURE")
# pylint:disable=abstract-method
class AzureFileSystem(ObjectFileSystem):
protocol = "azure"
PARAM_CHECKSUM = "etag"
REQUIRES = {
"adlfs": "adlfs",
"knack": "knack",
"azure-identity": "azure.identity",
}
@classmethod
def _strip_protocol(cls, path: str):
opts = infer_storage_options(path)
if opts.get("host"):
return "{host}{path}".format(**opts)
return _az_config().get("storage", "container_name", None)
def unstrip_protocol(self, path: str) -> str:
return "azure://" + path.lstrip("/")
@staticmethod
def _get_kwargs_from_urls(urlpath):
ops = infer_storage_options(urlpath)
if "host" in ops:
return {"bucket": ops["host"]}
return {}
def _prepare_credentials(self, **config):
from azure.identity.aio import DefaultAzureCredential
# Disable spam from failed cred types for DefaultAzureCredential
logging.getLogger("azure.identity.aio").setLevel(logging.ERROR)
login_info = {}
login_info["connection_string"] = config.get(
"connection_string",
_az_config().get("storage", "connection_string", None),
)
login_info["account_name"] = config.get(
"account_name", _az_config().get("storage", "account", None)
)
login_info["account_key"] = config.get(
"account_key", _az_config().get("storage", "key", None)
)
login_info["sas_token"] = config.get(
"sas_token", _az_config().get("storage", "sas_token", None)
)
login_info["tenant_id"] = config.get("tenant_id")
login_info["client_id"] = config.get("client_id")
login_info["client_secret"] = config.get("client_secret")
if not (login_info["account_name"] or login_info["connection_string"]):
raise AzureAuthError(
"Authentication to Azure Blob Storage requires either "
"account_name or connection_string."
)
any_secondary = any(
value for key, value in login_info.items() if key != "account_name"
)
if (
login_info["account_name"]
and not any_secondary
and not config.get("allow_anonymous_login", False)
):
with fsspec_loop():
login_info["credential"] = DefaultAzureCredential(
exclude_interactive_browser_credential=False,
exclude_environment_credential=config.get(
"exclude_environment_credential", False
),
exclude_visual_studio_code_credential=config.get(
"exclude_visual_studio_code_credential", False
),
exclude_shared_token_cache_credential=config.get(
"exclude_shared_token_cache_credential", False
),
exclude_managed_identity_credential=config.get(
"exclude_managed_identity_credential", False
),
)
for login_method, required_keys in [ # noqa
("connection string", ["connection_string"]),
(
"AD service principal",
["tenant_id", "client_id", "client_secret"],
),
("account key", ["account_name", "account_key"]),
("SAS token", ["account_name", "sas_token"]),
(
f"default credentials ({_DEFAULT_CREDS_STEPS})",
["account_name", "credential"],
),
("anonymous login", ["account_name"]),
]:
if all(login_info.get(key) is not None for key in required_keys):
break
else:
login_method = None
self.login_method = login_method
return login_info
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
from adlfs import AzureBlobFileSystem
from azure.core.exceptions import AzureError
try:
return AzureBlobFileSystem(**self.fs_args)
except (ValueError, AzureError) as e:
raise AzureAuthError(
f"Authentication to Azure Blob Storage via {self.login_method}"
" failed."
) from e
-243
View File
@@ -1,243 +0,0 @@
import logging
import os
import threading
from funcy import cached_property, wrap_prop
from ..base import FileSystem
from ..errors import AuthError, ConfigError
from ..utils import tmp_fname
logger = logging.getLogger(__name__)
FOLDER_MIME_TYPE = "application/vnd.google-apps.folder"
class GDriveAuthError(AuthError):
def __init__(self, cred_location):
if cred_location:
message = (
"GDrive remote auth failed with credentials in '{}'.\n"
"Backup first, remove or fix them, and run again.\n"
"It should do auth again and refresh the credentials.\n\n"
"Details:".format(cred_location)
)
else:
message = "Failed to authenticate GDrive remote"
super().__init__(message)
class GDriveFileSystem(FileSystem): # pylint:disable=abstract-method
protocol = "gdrive"
PARAM_CHECKSUM = "checksum"
REQUIRES = {"pydrive2": "pydrive2"}
# Always prefer traverse for GDrive since API usage quotas are a concern.
TRAVERSE_WEIGHT_MULTIPLIER = 1
GDRIVE_CREDENTIALS_DATA = "GDRIVE_CREDENTIALS_DATA"
DEFAULT_USER_CREDENTIALS_FILE = "gdrive-user-credentials.json"
DEFAULT_GDRIVE_CLIENT_ID = "710796635688-iivsgbgsb6uv1fap6635dhvuei09o66c.apps.googleusercontent.com" # noqa: E501
DEFAULT_GDRIVE_CLIENT_SECRET = "a1Fz59uTpVNeG_VGuSKDLJXv"
def __init__(self, **config):
from fsspec.utils import infer_storage_options
super().__init__(**config)
self.url = config["url"]
opts = infer_storage_options(self.url)
if not opts["host"]:
raise ConfigError("Empty GDrive URL '{}'.".format(config["url"]))
self._bucket = opts["host"]
self._path = opts["path"].lstrip("/")
self._trash_only = config.get("gdrive_trash_only")
self._use_service_account = config.get("gdrive_use_service_account")
self._service_account_user_email = config.get(
"gdrive_service_account_user_email"
)
self._service_account_json_file_path = config.get(
"gdrive_service_account_json_file_path"
)
self._client_id = config.get("gdrive_client_id")
self._client_secret = config.get("gdrive_client_secret")
self._validate_config()
tmp_dir = config["gdrive_credentials_tmp_dir"]
assert tmp_dir
self._gdrive_service_credentials_path = tmp_fname(
os.path.join(tmp_dir, "")
)
self._gdrive_user_credentials_path = (
tmp_fname(os.path.join(tmp_dir, ""))
if os.getenv(GDriveFileSystem.GDRIVE_CREDENTIALS_DATA)
else config.get(
"gdrive_user_credentials_file",
os.path.join(tmp_dir, self.DEFAULT_USER_CREDENTIALS_FILE),
)
)
@classmethod
def _strip_protocol(cls, path):
from fsspec.utils import infer_storage_options
opts = infer_storage_options(path)
return "{host}{path}".format(**opts)
def unstrip_protocol(self, path):
return f"gdrive://{path}"
@staticmethod
def _get_kwargs_from_urls(urlpath):
return {"url": urlpath}
def _validate_config(self):
# Validate Service Account configuration
if (
self._use_service_account
and not self._service_account_json_file_path
and GDriveFileSystem.GDRIVE_CREDENTIALS_DATA not in os.environ
):
raise ConfigError(
"To use service account, set "
"`gdrive_service_account_json_file_path`, and optionally"
"`gdrive_service_account_user_email` in DVC config."
)
# Validate OAuth 2.0 Client ID configuration
if not self._use_service_account:
if bool(self._client_id) != bool(self._client_secret):
raise ConfigError(
"Please specify GDrive's client ID and secret in "
"DVC config or omit both to use the defaults."
)
@cached_property
def credentials_location(self):
"""
Helper to determine where will GDrive remote read credentials from.
Useful for tests, exception messages, etc. Returns either env variable
name if it's set or actual path to the credentials file.
"""
if os.getenv(GDriveFileSystem.GDRIVE_CREDENTIALS_DATA):
return GDriveFileSystem.GDRIVE_CREDENTIALS_DATA
if os.path.exists(self._gdrive_user_credentials_path):
return self._gdrive_user_credentials_path
return None
@staticmethod
def _validate_credentials(auth, settings):
"""
Detects discrepancy in DVC config and cached credentials file.
Usually happens when a second remote is added and it is using
the same credentials default file. Or when someones decides to change
DVC config client id or secret but forgets to remove the cached
credentials file.
"""
if not os.getenv(GDriveFileSystem.GDRIVE_CREDENTIALS_DATA):
if (
settings["client_config"]["client_id"]
!= auth.credentials.client_id
or settings["client_config"]["client_secret"]
!= auth.credentials.client_secret
):
logger.warning(
"Client ID and secret configured do not match the "
"actual ones used\nto access the remote. Do you "
"use multiple GDrive remotes and forgot to\nset "
"`gdrive_user_credentials_file` for one or more of them?"
)
@wrap_prop(threading.RLock())
@cached_property
def fs(self):
from pydrive2.auth import GoogleAuth
from pydrive2.fs import GDriveFileSystem as _GDriveFileSystem
temporary_save_path = self._gdrive_user_credentials_path
is_credentials_temp = os.getenv(
GDriveFileSystem.GDRIVE_CREDENTIALS_DATA
)
if self._use_service_account:
temporary_save_path = self._gdrive_service_credentials_path
if is_credentials_temp:
with open(temporary_save_path, "w", encoding="utf-8") as cred_file:
cred_file.write(
os.getenv(GDriveFileSystem.GDRIVE_CREDENTIALS_DATA)
)
auth_settings = {
"client_config_backend": "settings",
"save_credentials": True,
"save_credentials_backend": "file",
"save_credentials_file": self._gdrive_user_credentials_path,
"get_refresh_token": True,
"oauth_scope": [
"https://www.googleapis.com/auth/drive",
"https://www.googleapis.com/auth/drive.appdata",
],
}
if self._use_service_account:
auth_settings["service_config"] = {
"client_user_email": self._service_account_user_email,
"client_json_file_path": self._service_account_json_file_path,
}
if is_credentials_temp:
auth_settings["service_config"][
"client_json_file_path"
] = temporary_save_path
else:
auth_settings["client_config"] = {
"client_id": self._client_id or self.DEFAULT_GDRIVE_CLIENT_ID,
"client_secret": self._client_secret
or self.DEFAULT_GDRIVE_CLIENT_SECRET,
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"token_uri": "https://oauth2.googleapis.com/token",
"revoke_uri": "https://oauth2.googleapis.com/revoke",
"redirect_uri": "",
}
GoogleAuth.DEFAULT_SETTINGS.update(auth_settings)
# Pass non existent settings path to force DEFAULT_SETTINGS loadings
gauth = GoogleAuth(settings_file="")
try:
logger.debug(
"GDrive remote auth with config '{}'.".format(
GoogleAuth.DEFAULT_SETTINGS
)
)
if self._use_service_account:
gauth.ServiceAuth()
else:
gauth.CommandLineAuth()
GDriveFileSystem._validate_credentials(gauth, auth_settings)
# Handle AuthenticationError, RefreshError and other auth failures
# It's hard to come up with a narrow exception, since PyDrive throws
# a lot of different errors - broken credentials file, refresh token
# expired, flow failed, etc.
except Exception as exc:
raise GDriveAuthError(self.credentials_location) from exc
finally:
if is_credentials_temp:
os.remove(temporary_save_path)
return _GDriveFileSystem(
f"{self._bucket}/{self._path}",
gauth,
trash_only=self._trash_only,
)
def upload_fobj(self, fobj, to_info, **kwargs):
self.makedirs(os.path.dirname(to_info))
return self.fs.upload_fobj(fobj, to_info, **kwargs)
-25
View File
@@ -1,25 +0,0 @@
import threading
from funcy import cached_property, wrap_prop
# pylint:disable=abstract-method
from ..base import ObjectFileSystem
class GSFileSystem(ObjectFileSystem):
protocol = "gs"
REQUIRES = {"gcsfs": "gcsfs"}
PARAM_CHECKSUM = "etag"
def _prepare_credentials(self, **config):
login_info = {"consistency": None}
login_info["project"] = config.get("projectname")
login_info["token"] = config.get("credentialpath")
return login_info
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
from gcsfs import GCSFileSystem
return GCSFileSystem(**self.fs_args)
-109
View File
@@ -1,109 +0,0 @@
import os
import re
import subprocess
import sys
import threading
from funcy import cached_property, wrap_prop
from ..base import FileSystem
CHECKSUM_REGEX = re.compile(r".*\t.*\t(?P<checksum>.*)")
def fix_env():
env = os.environ.copy()
if getattr(sys, "frozen", False):
lp_key = "LD_LIBRARY_PATH"
lp_orig = env.get(lp_key + "_ORIG", None)
if lp_orig is not None:
env[lp_key] = lp_orig
else:
env.pop(lp_key, None)
return env
# pylint: disable=abstract-method
class HDFSFileSystem(FileSystem):
protocol = "hdfs"
REQUIRES = {"fsspec": "fsspec", "pyarrow": "pyarrow"}
PARAM_CHECKSUM = "checksum"
@classmethod
def _strip_protocol(cls, path: str) -> str:
from fsspec.utils import infer_storage_options
return infer_storage_options(path)["path"]
def unstrip_protocol(self, path: str) -> str:
host = self.fs_args["host"]
port = self.fs_args.get("port")
netloc = host + (f":{port}" if port else "")
return "hdfs://" + netloc + "/" + path.lstrip("/")
@staticmethod
def _get_kwargs_from_urls(urlpath):
from fsspec.implementations.arrow import HadoopFileSystem
# pylint:disable=protected-access
return HadoopFileSystem._get_kwargs_from_urls(urlpath)
def _prepare_credentials(self, **config):
return config
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
from fsspec.implementations.arrow import HadoopFileSystem
return HadoopFileSystem(**self.fs_args)
def checksum(self, path):
return self._checksum(path)
def _checksum(self, path, **kwargs):
# PyArrow doesn't natively support retrieving the
# checksum, so we have to use hadoop fs
url = self.unstrip_protocol(path)
result = self._run_command(
f"checksum {url}",
env=fix_env(),
user=self.fs_args.get("user"),
)
if result is None:
return None
match = CHECKSUM_REGEX.match(result)
if match is None:
return None
return match.group("checksum")
def _run_command(self, cmd, env=None, user=None):
cmd = "hadoop fs -" + cmd
if user:
cmd = f"HADOOP_USER_NAME={user} " + cmd
# NOTE: close_fds doesn't work with redirected stdin/stdout/stderr.
# See https://github.com/iterative/dvc/issues/1197.
close_fds = os.name != "nt"
executable = os.getenv("SHELL") if os.name != "nt" else None
p = subprocess.Popen(
cmd,
shell=True,
close_fds=close_fds,
executable=executable,
env=env or os.environ,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
out, err = p.communicate()
if p.returncode != 0:
raise subprocess.CalledProcessError(p.returncode, cmd, out, err)
else:
return out.decode("utf-8")
-166
View File
@@ -1,166 +0,0 @@
import threading
from getpass import getpass
from typing import BinaryIO, Union
from funcy import cached_property, memoize, wrap_with
from ..base import AnyFSPath, FileSystem
from ..callbacks import DEFAULT_CALLBACK, Callback
from ..errors import ConfigError
@wrap_with(threading.Lock())
@memoize
def ask_password(host, user):
return getpass(f"Enter a password for host '{host}' user '{user}':\n")
def make_context(ssl_verify):
if isinstance(ssl_verify, bool) or ssl_verify is None:
return ssl_verify
# If this is a path, then we will create an
# SSL context for it, and load the given certificate.
import ssl
context = ssl.create_default_context()
context.load_verify_locations(ssl_verify)
return context
# pylint: disable=abstract-method
class HTTPFileSystem(FileSystem):
protocol = "http"
PARAM_CHECKSUM = "checksum"
REQUIRES = {"aiohttp": "aiohttp", "aiohttp-retry": "aiohttp_retry"}
CAN_TRAVERSE = False
SESSION_RETRIES = 5
SESSION_BACKOFF_FACTOR = 0.1
REQUEST_TIMEOUT = 60
def _prepare_credentials(self, **config):
import aiohttp
from fsspec.asyn import fsspec_loop
credentials = {}
client_kwargs = credentials.setdefault("client_kwargs", {})
if config.get("auth"):
user = config.get("user")
password = config.get("password")
custom_auth_header = config.get("custom_auth_header")
if password is None and config.get("ask_password"):
password = ask_password(config.get("url"), user or "custom")
auth_method = config["auth"]
if auth_method == "basic":
if user is None or password is None:
raise ConfigError(
"HTTP 'basic' authentication require both "
"'user' and 'password'"
)
client_kwargs["auth"] = aiohttp.BasicAuth(user, password)
elif auth_method == "custom":
if custom_auth_header is None or password is None:
raise ConfigError(
"HTTP 'custom' authentication require both "
"'custom_auth_header' and 'password'"
)
credentials["headers"] = {custom_auth_header: password}
else:
raise NotImplementedError(
f"Auth method {auth_method!r} is not supported."
)
# Force cleanup of closed SSL transports.
# https://github.com/iterative/dvc/issues/7414
connector_kwargs = {"enable_cleanup_closed": True}
if "ssl_verify" in config:
connector_kwargs.update(ssl=make_context(config["ssl_verify"]))
with fsspec_loop():
client_kwargs["connector"] = aiohttp.TCPConnector(
**connector_kwargs
)
# The connector should not be owned by aiohttp.ClientSession since
# it is closed by fsspec (HTTPFileSystem.close_session)
client_kwargs["connector_owner"] = False
# Allow reading proxy configurations from the environment.
client_kwargs["trust_env"] = True
credentials["get_client"] = self.get_client
self.upload_method = config.get("method", "POST")
return credentials
async def get_client(self, **kwargs):
import aiohttp
from aiohttp_retry import ExponentialRetry, RetryClient
kwargs["retry_options"] = ExponentialRetry(
attempts=self.SESSION_RETRIES,
factor=self.SESSION_BACKOFF_FACTOR,
max_timeout=self.REQUEST_TIMEOUT,
exceptions=[aiohttp.ClientError],
)
# The default timeout for the aiohttp is 300 seconds
# which is too low for DVC's interactions (especially
# on the read) when dealing with large data blobs. We
# unlimit the total time to read, and only limit the
# time that is spent when connecting to the remote server.
kwargs["timeout"] = aiohttp.ClientTimeout(
total=None,
connect=self.REQUEST_TIMEOUT,
sock_connect=self.REQUEST_TIMEOUT,
sock_read=self.REQUEST_TIMEOUT,
)
return RetryClient(**kwargs)
@cached_property
def fs(self):
from fsspec.implementations.http import (
HTTPFileSystem as _HTTPFileSystem,
)
return _HTTPFileSystem(**self.fs_args)
def unstrip_protocol(self, path: str) -> str:
return path
def put_file(
self,
from_file: Union[AnyFSPath, BinaryIO],
to_info: AnyFSPath,
callback: Callback = DEFAULT_CALLBACK,
size: int = None,
**kwargs,
) -> None:
kwargs.setdefault("method", self.upload_method)
super().put_file(
from_file, to_info, callback=callback, size=size, **kwargs
)
# pylint: disable=arguments-differ
def find(self, *args, **kwargs):
raise NotImplementedError
def isdir(self, *args, **kwargs):
return False
def isfile(self, *args, **kwargs):
return True
def ls(self, *args, **kwargs):
raise NotImplementedError
def walk(self, *args, **kwargs):
raise NotImplementedError
# pylint: enable=arguments-differ
-5
View File
@@ -1,5 +0,0 @@
from .http import HTTPFileSystem
class HTTPSFileSystem(HTTPFileSystem): # pylint:disable=abstract-method
protocol = "https"
-197
View File
@@ -1,197 +0,0 @@
import logging
import os
import shutil
import threading
from fsspec import AbstractFileSystem
from funcy import cached_property, wrap_prop
from .. import system
from ..base import FileSystem
from ..utils import copyfile, makedirs, move, remove, tmp_fname
logger = logging.getLogger(__name__)
# pylint:disable=abstract-method, arguments-differ
class FsspecLocalFileSystem(AbstractFileSystem):
sep = os.sep
def __init__(self, *args, **kwargs):
from fsspec.implementations.local import LocalFileSystem as LocalFS
super().__init__(*args, **kwargs)
self.fs = LocalFS()
def makedirs(self, path, exist_ok=False):
makedirs(path, exist_ok=exist_ok)
def mkdir(self, path, create_parents=True, **kwargs):
if self.exists(path):
raise FileExistsError(path)
if create_parents:
self.makedirs(path, exist_ok=True)
else:
os.mkdir(path, **kwargs)
def lexists(self, path, **kwargs):
return os.path.lexists(path)
def exists(self, path, **kwargs):
# TODO: replace this with os.path.exists once the problem is fixed on
# the fsspec https://github.com/intake/filesystem_spec/issues/742
return os.path.lexists(path)
def checksum(self, path) -> str:
from fsspec.utils import tokenize
st = os.stat(path)
return str(int(tokenize([st.st_ino, st.st_mtime, st.st_size]), 16))
def info(self, path, **kwargs):
return self.fs.info(path)
def ls(self, path, **kwargs):
return self.fs.ls(path, **kwargs)
def isfile(self, path) -> bool:
return os.path.isfile(path)
def isdir(self, path) -> bool:
return os.path.isdir(path)
def walk(self, path, maxdepth=None, topdown=True, **kwargs):
"""Directory fs generator.
See `os.walk` for the docs. Differences:
- no support for symlinks
"""
for root, dirs, files in os.walk(
path,
topdown=topdown,
):
yield os.path.normpath(root), dirs, files
def find(self, path, **kwargs):
for root, _, files in self.walk(path, **kwargs):
for file in files:
# NOTE: os.path.join is ~5.5 times slower
yield f"{root}{os.sep}{file}"
@classmethod
def _parent(cls, path):
return os.path.dirname(path)
def put_file(self, lpath, rpath, callback=None, **kwargs):
parent = self._parent(rpath)
makedirs(parent, exist_ok=True)
tmp_file = os.path.join(parent, tmp_fname())
copyfile(lpath, tmp_file, callback=callback)
os.replace(tmp_file, rpath)
def get_file(self, rpath, lpath, callback=None, **kwargs):
copyfile(rpath, lpath, callback=callback)
def mv(self, path1, path2, **kwargs):
self.makedirs(self._parent(path2), exist_ok=True)
move(path1, path2)
def rmdir(self, path):
os.rmdir(path)
def rm_file(self, path):
remove(path)
def rm(self, path, recursive=False, maxdepth=None):
remove(path)
def cp_file(self, path1, path2, **kwargs):
return self.copy(path1, path2, **kwargs)
def copy(self, path1, path2, recursive=False, on_error=None, **kwargs):
tmp_info = os.path.join(self._parent(path2), tmp_fname(""))
try:
copyfile(path1, tmp_info)
os.rename(tmp_info, path2)
except Exception:
self.rm_file(tmp_info)
raise
def open(self, path, mode="r", encoding=None, **kwargs):
return open(path, mode=mode, encoding=encoding)
def symlink(self, path1, path2):
return system.symlink(path1, path2)
@staticmethod
def is_symlink(path):
return system.is_symlink(path)
@staticmethod
def is_hardlink(path):
return system.is_hardlink(path)
def hardlink(self, path1, path2):
# If there are a lot of empty files (which happens a lot in datasets),
# and the cache type is `hardlink`, we might reach link limits and
# will get something like: `too many links error`
#
# This is because all those empty files will have the same hash
# (i.e. 68b329da9893e34099c7d8ad5cb9c940), therefore, they will be
# linked to the same file in the cache.
#
# From https://en.wikipedia.org/wiki/Hard_link
# * ext4 limits the number of hard links on a file to 65,000
# * Windows with NTFS has a limit of 1024 hard links on a file
#
# That's why we simply create an empty file rather than a link.
if self.size(path1) == 0:
self.open(path2, "w").close()
logger.debug("Created empty file: %s -> %s", path1, path2)
return
return system.hardlink(path1, path2)
def reflink(self, path1, path2):
return system.reflink(path1, path2)
def created(self, path):
return self.fs.created(path)
def modified(self, path):
return self.fs.modified(path)
class LocalFileSystem(FileSystem):
sep = os.sep
protocol = "local"
PARAM_CHECKSUM = "md5"
PARAM_PATH = "path"
TRAVERSE_PREFIX_LEN = 2
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
return FsspecLocalFileSystem(**self.config)
@cached_property
def path(self):
from ..path import Path
return Path(self.sep, getcwd=os.getcwd, realpath=os.path.realpath)
def upload_fobj(self, fobj, to_info, **kwargs):
self.makedirs(self.path.parent(to_info))
tmp_info = self.path.join(self.path.parent(to_info), tmp_fname(""))
try:
with open(tmp_info, "wb+") as fdest:
shutil.copyfileobj(fobj, fdest)
os.rename(tmp_info, to_info)
except Exception:
self.remove(tmp_info)
raise
localfs = LocalFileSystem()
-24
View File
@@ -1,24 +0,0 @@
import threading
from funcy import cached_property, wrap_prop
from ..base import FileSystem
class MemoryFileSystem(FileSystem): # pylint:disable=abstract-method
protocol = "memory"
PARAM_CHECKSUM = "md5"
def __eq__(self, other):
# NOTE: all fsspec MemoryFileSystem instances are equivalent and use a
# single global store
return isinstance(other, type(self))
__hash__ = FileSystem.__hash__
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
from fsspec.implementations.memory import MemoryFileSystem as MemFS
return MemFS(**self.fs_args)
-35
View File
@@ -1,35 +0,0 @@
import logging
import os
import threading
from funcy import cached_property, wrap_prop
from ..base import ObjectFileSystem
logger = logging.getLogger(__name__)
# pylint:disable=abstract-method
class OSSFileSystem(ObjectFileSystem):
protocol = "oss"
REQUIRES = {"ossfs": "ossfs"}
PARAM_CHECKSUM = "etag"
LIST_OBJECT_PAGE_SIZE = 100
def _prepare_credentials(self, **config):
login_info = {}
login_info["key"] = config.get("oss_key_id") or os.getenv(
"OSS_ACCESS_KEY_ID"
)
login_info["secret"] = config.get("oss_key_secret") or os.getenv(
"OSS_ACCESS_KEY_SECRET"
)
login_info["endpoint"] = config.get("oss_endpoint")
return login_info
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
from ossfs import OSSFileSystem as _OSSFileSystem
return _OSSFileSystem(**self.fs_args)
-157
View File
@@ -1,157 +0,0 @@
import os
import threading
from collections import defaultdict
from funcy import cached_property, wrap_prop
from ..base import ObjectFileSystem
from ..errors import ConfigError
from ..utils import flatten, unflatten
_AWS_CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".aws", "config")
# pylint:disable=abstract-method
class S3FileSystem(ObjectFileSystem):
protocol = "s3"
REQUIRES = {"s3fs": "s3fs", "boto3": "boto3"}
PARAM_CHECKSUM = "etag"
_GRANTS = {
"grant_full_control": "GrantFullControl",
"grant_read": "GrantRead",
"grant_read_acp": "GrantReadACP",
"grant_write_acp": "GrantWriteACP",
}
_TRANSFER_CONFIG_ALIASES = {
"max_queue_size": "max_io_queue",
"max_concurrent_requests": "max_concurrency",
"multipart_threshold": "multipart_threshold",
"multipart_chunksize": "multipart_chunksize",
}
def _split_s3_config(self, s3_config):
"""Splits the general s3 config into 2 different config
objects, one for transfer.TransferConfig and other is the
general session config"""
from boto3.s3.transfer import TransferConfig
from ..utils import human_readable_to_bytes
config, transfer_config = {}, {}
for key, value in s3_config.items():
if key in self._TRANSFER_CONFIG_ALIASES:
if key in {"multipart_chunksize", "multipart_threshold"}:
# cast human readable sizes (like 24MiB) to integers
value = human_readable_to_bytes(value)
else:
value = int(value)
transfer_config[self._TRANSFER_CONFIG_ALIASES[key]] = value
else:
config[key] = value
# pylint: disable=attribute-defined-outside-init
self._transfer_config = TransferConfig(**transfer_config)
return config
def _load_aws_config_file(self, profile):
from botocore.configloader import load_config
# pylint: disable=attribute-defined-outside-init
self._transfer_config = None
config_path = os.environ.get("AWS_CONFIG_FILE", _AWS_CONFIG_PATH)
if not os.path.exists(config_path):
return {}
config = load_config(config_path)
profile_config = config["profiles"].get(profile or "default")
if not profile_config:
return {}
s3_config = profile_config.get("s3", {})
return self._split_s3_config(s3_config)
def _prepare_credentials(self, **config):
login_info = defaultdict(dict)
# credentials
login_info["key"] = config.get("access_key_id")
login_info["secret"] = config.get("secret_access_key")
login_info["token"] = config.get("session_token")
# session configuration
login_info["profile"] = config.get("profile")
login_info["use_ssl"] = config.get("use_ssl", True)
# extra client configuration
client = login_info["client_kwargs"]
client["region_name"] = config.get("region")
client["endpoint_url"] = config.get("endpointurl")
client["verify"] = config.get("ssl_verify")
# timeout configuration
config_kwargs = login_info["config_kwargs"]
config_kwargs["read_timeout"] = config.get("read_timeout")
config_kwargs["connect_timeout"] = config.get("connect_timeout")
# encryptions
additional = login_info["s3_additional_kwargs"]
additional["ServerSideEncryption"] = config.get("sse")
additional["SSEKMSKeyId"] = config.get("sse_kms_key_id")
additional["ACL"] = config.get("acl")
for grant_option, grant_key in self._GRANTS.items():
if config.get(grant_option):
if additional["ACL"]:
raise ConfigError(
"`acl` and `grant_*` AWS S3 config options "
"are mutually exclusive"
)
additional[grant_key] = config[grant_option]
# config kwargs
session_config = login_info["config_kwargs"]
session_config["s3"] = self._load_aws_config_file(
login_info["profile"]
)
shared_creds = config.get("credentialpath")
if shared_creds:
os.environ.setdefault("AWS_SHARED_CREDENTIALS_FILE", shared_creds)
if (
client["region_name"] is None
and session_config["s3"].get("region_name") is None
and os.getenv("AWS_REGION") is None
):
# Enable bucket region caching
login_info["cache_regions"] = config.get("cache_regions", True)
config_path = config.get("configpath")
if config_path:
os.environ.setdefault("AWS_CONFIG_FILE", config_path)
return unflatten(
{
key: value
for key, value in flatten(login_info).items()
if value is not None
}
)
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
from s3fs import S3FileSystem as _S3FileSystem
return _S3FileSystem(**self.fs_args)
@classmethod
def _strip_protocol(cls, path: str) -> str:
from fsspec.utils import infer_storage_options
return infer_storage_options(path)["path"]
def unstrip_protocol(self, path):
return "s3://" + path.lstrip("/")
-135
View File
@@ -1,135 +0,0 @@
import getpass
import os.path
import threading
from funcy import cached_property, memoize, silent, wrap_prop, wrap_with
from ..base import FileSystem
from ..callbacks import DEFAULT_CALLBACK
from ..utils import as_atomic
DEFAULT_PORT = 22
@wrap_with(threading.Lock())
@memoize
def ask_password(host, user, port):
return getpass.getpass(
"Enter a private key passphrase or a password for "
f"host '{host}' port '{port}' user '{user}':\n"
)
# pylint:disable=abstract-method
class SSHFileSystem(FileSystem):
protocol = "ssh"
REQUIRES = {"sshfs": "sshfs"}
PARAM_CHECKSUM = "md5"
@classmethod
def _strip_protocol(cls, path: str) -> str:
from fsspec.utils import infer_storage_options
return infer_storage_options(path)["path"]
def unstrip_protocol(self, path: str) -> str:
host = self.fs_args["host"]
port = self.fs_args["port"]
path = path.lstrip("/")
return f"ssh://{host}:{port}/{path}"
def _prepare_credentials(self, **config):
self.CAN_TRAVERSE = True
from sshfs.config import parse_config
login_info = {}
try:
user_ssh_config = parse_config(host=config["host"])
except FileNotFoundError:
user_ssh_config = {}
login_info["host"] = user_ssh_config.get("Hostname", config["host"])
login_info["username"] = (
config.get("user")
or config.get("username")
or user_ssh_config.get("User")
or getpass.getuser()
)
login_info["port"] = (
config.get("port")
or silent(int)(user_ssh_config.get("Port"))
or DEFAULT_PORT
)
if config.get("ask_password") and config.get("password") is None:
config["password"] = ask_password(
login_info["host"], login_info["username"], login_info["port"]
)
login_info["password"] = config.get("password")
login_info["passphrase"] = config.get("password")
raw_keys = []
if config.get("keyfile"):
raw_keys.append(config.get("keyfile"))
elif user_ssh_config.get("IdentityFile"):
raw_keys.extend(user_ssh_config.get("IdentityFile"))
if raw_keys:
login_info["client_keys"] = [
os.path.expanduser(key) for key in raw_keys
]
login_info["timeout"] = config.get("timeout", 1800)
# These two settings fine tune the asyncssh to use the
# fastest encryption algorithm and disable compression
# altogether (since it blocking, it is slowing down
# the transfers in a considerable rate, and even for
# compressible data it is making it extremely slow).
# See: https://github.com/ronf/asyncssh/issues/374
login_info["encryption_algs"] = [
"[email protected]",
"aes256-ctr",
"aes192-ctr",
"aes128-ctr",
]
login_info["compression_algs"] = None
login_info["gss_auth"] = config.get("gss_auth", False)
login_info["agent_forwarding"] = config.get("agent_forwarding", True)
login_info["proxy_command"] = user_ssh_config.get("ProxyCommand")
# We are going to automatically add stuff to known_hosts
# something like paramiko's AutoAddPolicy()
login_info["known_hosts"] = None
return login_info
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
from sshfs import SSHFileSystem as _SSHFileSystem
return _SSHFileSystem(**self.fs_args)
# Ensure that if an interrupt happens during the transfer, we don't
# pollute the cache.
def upload_fobj(self, fobj, to_info, **kwargs):
with as_atomic(self, to_info) as tmp_file:
super().upload_fobj(fobj, tmp_file, **kwargs)
def put_file(
self,
from_file,
to_info,
callback=DEFAULT_CALLBACK,
size=None,
**kwargs,
):
with as_atomic(self, to_info) as tmp_file:
super().put_file(
from_file, tmp_file, callback=callback, size=size, **kwargs
)
-90
View File
@@ -1,90 +0,0 @@
import logging
import threading
from funcy import cached_property, wrap_prop
from ..base import FileSystem
from .http import ask_password
logger = logging.getLogger(__name__)
class WebDAVFileSystem(FileSystem): # pylint:disable=abstract-method
protocol = "webdav"
root_marker = ""
REQUIRES = {"webdav4": "webdav4"}
PARAM_CHECKSUM = "etag"
def __init__(self, **config):
super().__init__(**config)
cert_path = config.get("cert_path", None)
key_path = config.get("key_path", None)
self.prefix = config.get("prefix", "")
self.fs_args.update(
{
"base_url": config["url"],
"cert": cert_path if not key_path else (cert_path, key_path),
"verify": config.get("ssl_verify", True),
"timeout": config.get("timeout", 30),
}
)
def unstrip_protocol(self, path: str) -> str:
return self.fs_args["base_url"] + "/" + path
@staticmethod
def _get_kwargs_from_urls(urlpath):
from urllib.parse import urlparse, urlunparse
parsed = urlparse(urlpath)
scheme = parsed.scheme.replace("webdav", "http")
return {
"prefix": parsed.path.rstrip("/"),
"host": urlunparse((scheme, parsed.netloc, "", None, None, None)),
"url": urlunparse(
(
scheme,
parsed.netloc,
parsed.path.rstrip("/"),
None,
None,
None,
)
),
"user": parsed.username,
}
def _prepare_credentials(self, **config):
user = config.get("user", None)
password = config.get("password", None)
headers = {}
token = config.get("token")
auth = None
if token:
headers.update({"Authorization": f"Bearer {token}"})
elif user:
if not password and config.get("ask_password"):
password = ask_password(config["host"], user)
auth = (user, password)
return {"headers": headers, "auth": auth}
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
from webdav4.fsspec import WebdavFileSystem
return WebdavFileSystem(**self.fs_args)
def upload_fobj(self, fobj, to_info, **kwargs):
size = kwargs.get("size")
# using upload_fileobj to directly upload fileobj rather than buffering
# and using overwrite=True to avoid check for an extra exists call,
# as caller should ensure that the file does not exist beforehand.
return self.fs.upload_fileobj(fobj, to_info, overwrite=True, size=size)
class WebDAVSFileSystem(WebDAVFileSystem): # pylint:disable=abstract-method
protocol = "webdavs"
-54
View File
@@ -1,54 +0,0 @@
import threading
from funcy import cached_property, wrap_prop
# pylint:disable=abstract-method
from ..base import FileSystem
class WebHDFSFileSystem(FileSystem):
protocol = "webhdfs"
REQUIRES = {"fsspec": "fsspec"}
PARAM_CHECKSUM = "checksum"
@classmethod
def _strip_protocol(cls, path: str) -> str:
from fsspec.utils import infer_storage_options
return infer_storage_options(path)["path"]
def unstrip_protocol(self, path: str) -> str:
host = self.fs_args["host"]
port = self.fs_args["port"]
path = path.lstrip("/")
return f"webhdfs://{host}:{port}/{path}"
@staticmethod
def _get_kwargs_from_urls(urlpath):
from fsspec.implementations.webhdfs import WebHDFS
return (
WebHDFS._get_kwargs_from_urls( # pylint:disable=protected-access
urlpath
)
)
def _prepare_credentials(self, **config):
self._ssl_verify = config.pop("ssl_verify", True)
principal = config.pop("kerberos_principal", None)
if principal:
config["kerb_kwargs"] = {"principal": principal}
return config
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
from fsspec.implementations.webhdfs import WebHDFS
fs = WebHDFS(**self.fs_args)
fs.session.verify = self._ssl_verify
return fs
def checksum(self, path):
ukey = self.fs.ukey(path)
return ukey["bytes"]
-131
View File
@@ -1,131 +0,0 @@
import ntpath
import posixpath
class Path:
def __init__(self, sep, getcwd=None, realpath=None):
def _getcwd():
return ""
self.getcwd = getcwd or _getcwd
self.realpath = realpath or self.abspath
if sep == posixpath.sep:
self.flavour = posixpath
elif sep == ntpath.sep:
self.flavour = ntpath
else:
raise ValueError(f"unsupported separator '{sep}'")
def chdir(self, path):
def _getcwd():
return path
self.getcwd = _getcwd
def join(self, *parts):
return self.flavour.join(*parts)
def split(self, path):
return self.flavour.split(path)
def splitext(self, path):
return self.flavour.splitext(path)
def normpath(self, path):
return self.flavour.normpath(path)
def isabs(self, path):
return self.flavour.isabs(path)
def abspath(self, path):
if not self.isabs(path):
path = self.join(self.getcwd(), path)
return self.normpath(path)
def commonprefix(self, path):
return self.flavour.commonprefix(path)
def commonpath(self, paths):
return self.flavour.commonpath(paths)
def parts(self, path):
drive, path = self.flavour.splitdrive(path.rstrip(self.flavour.sep))
ret = []
while True:
path, part = self.flavour.split(path)
if part:
ret.append(part)
continue
if path:
ret.append(path)
break
ret.reverse()
if drive:
ret = [drive] + ret
return tuple(ret)
def parent(self, path):
return self.flavour.dirname(path)
def dirname(self, path):
return self.parent(path)
def parents(self, path):
while True:
parent = self.flavour.dirname(path)
if parent == path:
break
yield parent
path = parent
def name(self, path):
return self.flavour.basename(path)
def suffix(self, path):
name = self.name(path)
_, dot, suffix = name.partition(".")
return dot + suffix
def with_name(self, path, name):
return self.join(self.parent(path), name)
def with_suffix(self, path, suffix):
return self.splitext(path)[0] + suffix
def isin(self, left, right):
if left == right:
return False
try:
common = self.commonpath([left, right])
except ValueError:
# Paths don't have the same drive
return False
return common == right
def isin_or_eq(self, left, right):
return left == right or self.isin(left, right)
def overlaps(self, left, right):
# pylint: disable=arguments-out-of-order
return self.isin_or_eq(left, right) or self.isin(right, left)
def relpath(self, path, start=None):
if start is None:
start = "."
return self.flavour.relpath(
self.abspath(path), start=self.abspath(start)
)
def relparts(self, path, start=None):
return self.parts(self.relpath(path, start=start))
def as_posix(self, path):
return path.replace(self.flavour.sep, posixpath.sep)
-15
View File
@@ -1,15 +0,0 @@
class Schemes:
SSH = "ssh"
HDFS = "hdfs"
WEBHDFS = "webhdfs"
S3 = "s3"
AZURE = "azure"
HTTP = "http"
HTTPS = "https"
GS = "gs"
GDRIVE = "gdrive"
LOCAL = "local"
OSS = "oss"
WEBDAV = "webdav"
WEBDAVS = "webdavs"
MEMORY = "memory"
-140
View File
@@ -1,140 +0,0 @@
import errno
import logging
import os
import platform
import stat
import sys
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .base import AnyFSPath
logger = logging.getLogger(__name__)
if os.name == "nt" and sys.version_info < (3, 8):
# NOTE: using backports for `os.path.realpath`
# See https://bugs.python.org/issue9949 for more info.
# pylint: disable=import-error, no-name-in-module
from jaraco.windows.filesystem.backports import realpath as _realpath
def realpath(path):
return _realpath(os.fspath(path))
else:
realpath = os.path.realpath
umask = os.umask(0)
os.umask(umask)
def hardlink(source: "AnyFSPath", link_name: "AnyFSPath") -> None:
# NOTE: we should really be using `os.link()` here with
# `follow_symlinks=True`, but unfortunately the implementation is
# buggy across platforms, so until it is fixed, we just dereference
# the symlink ourselves here.
#
# See https://bugs.python.org/issue41355 for more info.
st = os.lstat(source)
if stat.S_ISLNK(st.st_mode):
src = realpath(source)
else:
src = source
os.link(src, link_name)
def symlink(source: "AnyFSPath", link_name: "AnyFSPath") -> None:
os.symlink(source, link_name)
def _reflink_darwin(src: "AnyFSPath", dst: "AnyFSPath") -> None:
import ctypes
def _cdll(name):
return ctypes.CDLL(name, use_errno=True)
LIBC = "libc.dylib"
LIBC_FALLBACK = "/usr/lib/libSystem.dylib"
try:
clib = _cdll(LIBC)
except OSError as exc:
logger.debug(
"unable to access '{}' (errno '{}'). "
"Falling back to '{}'.".format(LIBC, exc.errno, LIBC_FALLBACK)
)
if exc.errno != errno.ENOENT:
raise
# NOTE: trying to bypass System Integrity Protection (SIP)
clib = _cdll(LIBC_FALLBACK)
if not hasattr(clib, "clonefile"):
raise OSError(
errno.ENOTSUP,
"'clonefile' not supported by the standard library",
)
clonefile = clib.clonefile
clonefile.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int]
clonefile.restype = ctypes.c_int
ret = clonefile(
ctypes.c_char_p(os.fsencode(src)),
ctypes.c_char_p(os.fsencode(dst)),
ctypes.c_int(0),
)
if ret:
err = ctypes.get_errno()
msg = os.strerror(err)
raise OSError(err, msg)
def _reflink_linux(src: "AnyFSPath", dst: "AnyFSPath") -> None:
import fcntl # pylint: disable=import-error
from contextlib import suppress
FICLONE = 0x40049409
try:
with open(src, "rb") as s, open(dst, "wb+") as d:
fcntl.ioctl(d.fileno(), FICLONE, s.fileno())
except OSError:
with suppress(OSError):
os.unlink(dst)
raise
def reflink(source: "AnyFSPath", link_name: "AnyFSPath") -> None:
source, link_name = os.fspath(source), os.fspath(link_name)
system = platform.system()
if system == "Darwin":
_reflink_darwin(source, link_name)
elif system == "Linux":
_reflink_linux(source, link_name)
else:
raise OSError(
errno.ENOTSUP,
f"reflink is not supported on '{system}'",
)
# NOTE: reflink has a new inode, but has the same mode as the src,
# so we need to chmod it to look like a normal copy.
os.chmod(link_name, 0o666 & ~umask)
def inode(path: "AnyFSPath") -> int:
ino = os.lstat(path).st_ino
logger.trace( # type: ignore[attr-defined]
"Path '%s' inode '%d'", path, ino
)
return ino
def is_symlink(path: "AnyFSPath") -> bool:
return os.path.islink(path)
def is_hardlink(path: "AnyFSPath") -> bool:
return os.stat(path).st_nlink > 1
-250
View File
@@ -1,250 +0,0 @@
import errno
import logging
import os
import shutil
import stat
import sys
from contextlib import contextmanager, suppress
from typing import TYPE_CHECKING, Iterator
from . import system
if TYPE_CHECKING:
from .base import AnyFSPath, FileSystem
from .callbacks import Callback
logger = logging.getLogger(__name__)
LOCAL_CHUNK_SIZE = 2**20 # 1 MB
def is_exec(mode: int) -> bool:
return bool(mode & (stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH))
def relpath(path: "AnyFSPath", start: "AnyFSPath" = os.curdir) -> "AnyFSPath":
path = os.fspath(path)
start = os.path.abspath(os.fspath(start))
# Windows path on different drive than curdir doesn't have relpath
if os.name == "nt":
# Since python 3.8 os.realpath resolves network shares to their UNC
# path. So, to be certain that relative paths correctly captured,
# we need to resolve to UNC path first. We resolve only the drive
# name so that we don't follow any 'real' symlinks on the path
def resolve_network_drive_windows(path_to_resolve):
drive, tail = os.path.splitdrive(path_to_resolve)
return os.path.join(os.path.realpath(drive), tail)
path = resolve_network_drive_windows(os.path.abspath(path))
start = resolve_network_drive_windows(start)
if not os.path.commonprefix([start, path]):
return path
return os.path.relpath(path, start)
def move(src: "AnyFSPath", dst: "AnyFSPath") -> None:
"""Atomically move src to dst and chmod it with mode.
Moving is performed in two stages to make the whole operation atomic in
case src and dst are on different filesystems and actual physical copying
of data is happening.
"""
from shortuuid import uuid
dst = os.path.abspath(dst)
tmp = f"{dst}.{uuid()}"
if os.path.islink(src):
shutil.copy(src, tmp)
_unlink(src, _chmod)
else:
shutil.move(src, tmp)
shutil.move(tmp, dst)
def _chmod(func, p, excinfo): # pylint: disable=unused-argument
perm = os.lstat(p).st_mode
perm |= stat.S_IWRITE
try:
os.chmod(p, perm)
except OSError as exc:
# broken symlink or file is not owned by us
if exc.errno not in [errno.ENOENT, errno.EPERM]:
raise
func(p)
def _unlink(path: "AnyFSPath", onerror):
try:
os.unlink(path)
except OSError:
onerror(os.unlink, path, sys.exc_info())
def remove(path: "AnyFSPath") -> None:
logger.debug("Removing '%s'", path)
try:
if os.path.isdir(path):
shutil.rmtree(path, onerror=_chmod)
else:
_unlink(path, _chmod)
except OSError as exc:
if exc.errno != errno.ENOENT:
raise
def makedirs(path, exist_ok: bool = False, mode: int = None) -> None:
if mode is None:
os.makedirs(path, exist_ok=exist_ok)
return
# Modified version of os.makedirs() with support for extended mode
# (e.g. S_ISGID)
head, tail = os.path.split(path)
if not tail:
head, tail = os.path.split(head)
if head and tail and not os.path.exists(head):
try:
makedirs(head, exist_ok=exist_ok, mode=mode)
except FileExistsError:
# Defeats race condition when another thread created the path
pass
cdir = os.curdir
if isinstance(tail, bytes):
cdir = bytes(os.curdir, "ASCII") # type: ignore[assignment]
if tail == cdir: # xxx/newdir/. exists if xxx/newdir exists
return
try:
os.mkdir(path, mode)
except OSError:
# Cannot rely on checking for EEXIST, since the operating system
# could give priority to other errors like EACCES or EROFS
if not exist_ok or not os.path.isdir(path):
raise
try:
os.chmod(path, mode)
except OSError:
logger.trace( # type: ignore[attr-defined]
"failed to chmod '%o' '%s'", mode, path, exc_info=True
)
def copyfile(
src: "AnyFSPath",
dest: "AnyFSPath",
callback: "Callback" = None,
no_progress_bar: bool = False,
name: str = None,
) -> None:
"""Copy file with progress bar"""
name = name if name else os.path.basename(dest)
total = os.stat(src).st_size
if os.path.isdir(dest):
dest = os.path.join(dest, os.path.basename(src))
if callback:
callback.set_size(total)
try:
system.reflink(src, dest)
except OSError:
from .callbacks import Callback
with open(src, "rb") as fsrc, open(dest, "wb+") as fdest:
with Callback.as_tqdm_callback(
callback,
size=total,
bytes=True,
disable=no_progress_bar,
desc=name,
) as cb:
wrapped = cb.wrap_attr(fdest, "write")
while True:
buf = fsrc.read(LOCAL_CHUNK_SIZE)
if not buf:
break
wrapped.write(buf)
if callback:
callback.absolute_update(total)
def walk_files(directory: "AnyFSPath") -> Iterator["AnyFSPath"]:
for root, _, files in os.walk(directory):
for f in files:
yield os.path.join(root, f)
def tmp_fname(fname: "AnyFSPath" = "") -> "AnyFSPath":
"""Temporary name for a partial download"""
from shortuuid import uuid
return os.fspath(fname) + "." + uuid() + ".tmp"
@contextmanager
def as_atomic(
fs: "FileSystem", to_info: "AnyFSPath", create_parents: bool = False
) -> Iterator["AnyFSPath"]:
parent = fs.path.parent(to_info)
if create_parents:
fs.makedirs(parent, exist_ok=True)
tmp_info = fs.path.join(parent, tmp_fname())
try:
yield tmp_info
except BaseException:
# Handle stuff like KeyboardInterrupt
# as well as other errors that might
# arise during file transfer.
with suppress(FileNotFoundError):
fs.remove(tmp_info)
raise
else:
fs.move(tmp_info, to_info)
# https://github.com/aws/aws-cli/blob/5aa599949f60b6af554fd5714d7161aa272716f7/awscli/customizations/s3/utils.py
MULTIPLIERS = {
"kb": 1024,
"mb": 1024**2,
"gb": 1024**3,
"tb": 1024**4,
"kib": 1024,
"mib": 1024**2,
"gib": 1024**3,
"tib": 1024**4,
}
def human_readable_to_bytes(value: str) -> int:
value = value.lower()
suffix = ""
if value.endswith(tuple(MULTIPLIERS.keys())):
size = 2
size += value[-2] == "i" # KiB, MiB etc
value, suffix = value[:-size], value[-size:]
multiplier = MULTIPLIERS.get(suffix, 1)
return int(value) * multiplier
def flatten(d):
import flatten_dict
return flatten_dict.flatten(d, reducer="dot")
def unflatten(d):
import flatten_dict
return flatten_dict.unflatten(d, splitter="dot")
-179
View File
@@ -1,179 +0,0 @@
import hashlib
import io
import logging
from typing import TYPE_CHECKING, Any, BinaryIO, Dict, Optional, Tuple
from .fs.callbacks import DEFAULT_CALLBACK, Callback, TqdmCallback
from .fs.implementations.local import localfs
from .fs.utils import is_exec
from .hash_info import HashInfo
from .istextfile import DEFAULT_CHUNK_SIZE, istextblock
from .meta import Meta
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from .fs.base import AnyFSPath, FileSystem
from .state import StateBase
def dos2unix(data: bytes) -> bytes:
return data.replace(b"\r\n", b"\n")
def get_hasher(name: str) -> "hashlib._Hash":
try:
return getattr(hashlib, name)()
except AttributeError:
return hashlib.new(name)
class HashStreamFile(io.IOBase):
def __init__(
self,
fobj: BinaryIO,
hash_name: str = "md5",
text: Optional[bool] = None,
) -> None:
self.fobj = fobj
self.total_read = 0
self.hasher = get_hasher(hash_name)
self.is_text: Optional[bool] = text
super().__init__()
def readable(self) -> bool:
return True
def tell(self) -> int:
return self.fobj.tell()
def read(self, n=-1) -> bytes:
chunk = self.fobj.read(n)
if self.is_text is None and chunk:
# do we need to buffer till the DEFAULT_CHUNK_SIZE?
self.is_text = istextblock(chunk[:DEFAULT_CHUNK_SIZE])
data = dos2unix(chunk) if self.is_text else chunk
self.hasher.update(data)
self.total_read += len(data)
return chunk
@property
def hash_value(self) -> str:
return self.hasher.hexdigest()
@property
def hash_name(self) -> str:
return self.hasher.name
def fobj_md5(
fobj: BinaryIO, chunk_size: int = 2**20, text: Optional[bool] = None
) -> str:
# ideally, we want the heuristics to be applied in a similar way,
# regardless of the size of the first chunk,
# for which we may need to buffer till DEFAULT_CHUNK_SIZE.
assert chunk_size >= DEFAULT_CHUNK_SIZE
stream = HashStreamFile(fobj, text=text)
while True:
data = stream.read(chunk_size)
if not data:
break
return stream.hash_value
def file_md5(
fname: "AnyFSPath",
fs: "FileSystem" = localfs,
callback: "Callback" = DEFAULT_CALLBACK,
text: Optional[bool] = None,
) -> str:
size = fs.size(fname) or 0
callback.set_size(size)
with fs.open(fname, "rb") as fobj:
return fobj_md5(callback.wrap_attr(fobj), text=text)
def _adapt_info(info: Dict[str, Any], scheme: str) -> Dict[str, Any]:
if scheme == "s3" and "ETag" in info:
info["etag"] = info["ETag"].strip('"')
elif scheme == "gs" and "etag" in info:
import base64
info["etag"] = base64.b64decode(info["etag"]).hex()
elif scheme.startswith("http") and (
"ETag" in info or "Content-MD5" in info
):
info["checksum"] = info.get("ETag") or info.get("Content-MD5")
return info
def _hash_file(
fs_path: "AnyFSPath",
fs: "FileSystem",
name: str,
callback: "Callback" = DEFAULT_CALLBACK,
) -> Tuple["str", Dict[str, Any]]:
info = _adapt_info(fs.info(fs_path), fs.protocol)
if name in info:
assert not info[name].endswith(".dir")
return info[name], info
if hasattr(fs, name):
func = getattr(fs, name)
return func(fs_path), info
if name == "md5":
return file_md5(fs_path, fs, callback=callback), info
raise NotImplementedError
class LargeFileHashingCallback(TqdmCallback):
"""Callback that only shows progress bar if self.size > LARGE_FILE_SIZE."""
LARGE_FILE_SIZE = 2**30
def __init__(self, *args, **kwargs):
kwargs.setdefault("bytes", True)
super().__init__(*args, **kwargs)
self._logged = False
# TqdmCallback force renders progress bar on `set_size`.
set_size = Callback.set_size
def call(self, hook_name=None, **kwargs):
if self.size and self.size > self.LARGE_FILE_SIZE:
if not self._logged:
desc = self.progress_bar.desc
logger.info(
f"Computing md5 for a large file '{desc}'. "
"This is only done once."
)
self._logged = True
super().call()
def hash_file(
fs_path: "AnyFSPath",
fs: "FileSystem",
name: str,
state: "StateBase" = None,
callback: "Callback" = None,
) -> Tuple["Meta", "HashInfo"]:
if state:
meta, hash_info = state.get(fs_path, fs)
if hash_info:
return meta, hash_info
cb = callback or LargeFileHashingCallback(desc=fs_path)
with cb:
hash_value, info = _hash_file(fs_path, fs, name, callback=cb)
hash_info = HashInfo(name, hash_value)
if state:
assert ".dir" not in hash_info.value
state.save(fs_path, fs, hash_info)
meta = Meta(size=info["size"], isexec=is_exec(info.get("mode", 0)))
return meta, hash_info
-49
View File
@@ -1,49 +0,0 @@
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Optional
HASH_DIR_SUFFIX = ".dir"
@dataclass
class HashInfo:
name: Optional[str]
value: Optional[str]
obj_name: Optional[str] = field(default=None, compare=False)
def __bool__(self):
return bool(self.value)
def __str__(self):
return f"{self.name}: {self.value}"
def __hash__(self):
return hash((self.name, self.value))
@classmethod
def from_dict(cls, d):
if not d:
return cls(None, None)
((name, value),) = d.items()
return cls(name, value)
def to_dict(self):
ret = OrderedDict()
if not self:
return ret
ret[self.name] = self.value
return ret
@property
def isdir(self):
if not self:
return False
return self.value.endswith(HASH_DIR_SUFFIX)
def as_raw(self) -> "HashInfo":
assert self.value
return HashInfo(
self.name, self.value.rsplit(HASH_DIR_SUFFIX)[0], self.obj_name
)
-39
View File
@@ -1,39 +0,0 @@
"""Use heuristics to guess if it is a text file or a binary file."""
# Based on https://eli.thegreenplace.net/2011/10/19/
# perls-guess-if-file-is-text-or-binary-implemented-in-python
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .fs.base import AnyFSPath, FileSystem
TEXT_CHARS = bytes(range(32, 127)) + b"\n\r\t\f\b"
DEFAULT_CHUNK_SIZE = 512
def istextblock(block: bytes) -> bool:
if not block:
# An empty file is considered a valid text file
return True
if b"\x00" in block:
# Files with null bytes are binary
return False
# Use translate's 'deletechars' argument to efficiently remove all
# occurrences of TEXT_CHARS from the block
nontext = block.translate(None, TEXT_CHARS)
return float(len(nontext)) / len(block) <= 0.30
def istextfile(
fname: "AnyFSPath", fs: "FileSystem", blocksize: int = DEFAULT_CHUNK_SIZE
) -> bool:
"""Uses heuristics to guess whether the given file is text or binary,
by reading a single block of bytes from the file.
If more than 30% of the chars in the block are non-text, or there
are NUL ('\x00') bytes in the block, assume this is a binary file.
"""
with fs.open(fname, "rb") as fobj:
block = fobj.read(blocksize)
return istextblock(block)
-47
View File
@@ -1,47 +0,0 @@
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from .db import ObjectDB
from .file import HashFile
@dataclass
class Meta:
PARAM_SIZE = "size"
PARAM_NFILES = "nfiles"
PARAM_ISEXEC = "isexec"
size: Optional[int] = field(default=None)
nfiles: Optional[int] = field(default=None)
isexec: Optional[bool] = field(default=False)
obj: Optional["HashFile"] = field(default=None)
odb: Optional["ObjectDB"] = field(default=None)
remote: Optional[str] = field(default=None)
@classmethod
def from_dict(cls, d):
if not d:
return cls()
size = d.pop(cls.PARAM_SIZE, None)
nfiles = d.pop(cls.PARAM_NFILES, None)
isexec = d.pop(cls.PARAM_ISEXEC, False)
return cls(size=size, nfiles=nfiles, isexec=isexec)
def to_dict(self):
ret = OrderedDict()
if self.size is not None:
ret[self.PARAM_SIZE] = self.size
if self.nfiles is not None:
ret[self.PARAM_NFILES] = self.nfiles
if self.isexec:
ret[self.PARAM_ISEXEC] = self.isexec
return ret
-186
View File
@@ -1,186 +0,0 @@
"""Manages state database used for checksum caching."""
import logging
import os
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from .fs import LocalFileSystem
from .fs.system import inode as get_inode
from .fs.utils import relpath
from .hash_info import HashInfo
from .utils import get_mtime_and_size
if TYPE_CHECKING:
from ._ignore import Ignore
logger = logging.getLogger(__name__)
class StateBase(ABC):
@abstractmethod
def close(self):
pass
@abstractmethod
def save(self, path, fs, hash_info):
pass
@abstractmethod
def get(self, path, fs):
pass
@abstractmethod
def save_link(self, path, fs):
pass
class StateNoop(StateBase):
def close(self):
pass
def save(self, path, fs, hash_info):
pass
def get(self, path, fs): # pylint: disable=unused-argument
return None, None
def save_link(self, path, fs):
pass
class State(StateBase): # pylint: disable=too-many-instance-attributes
def __init__(self, root_dir=None, tmp_dir=None, ignore: "Ignore" = None):
from .cache import Cache
super().__init__()
self.tmp_dir = tmp_dir
self.root_dir = root_dir
self.ignore = ignore
if not tmp_dir:
return
links_dir = os.path.join(tmp_dir, "links")
md5s_dir = os.path.join(tmp_dir, "md5s")
self.links = Cache(links_dir, eviction_policy="least-recently-used")
self.md5s = Cache(md5s_dir, eviction_policy="least-recently-used")
def close(self):
self.md5s.close()
self.links.close()
def save(self, path, fs, hash_info):
"""Save hash for the specified path info.
Args:
path (str): path to save hash for.
hash_info (HashInfo): hash to save.
"""
if not isinstance(fs, LocalFileSystem):
return
mtime, size = get_mtime_and_size(path, fs, self.ignore)
inode = get_inode(path)
logger.debug(
"state save (%s, %s, %s) %s",
inode,
mtime,
str(size),
hash_info.value,
)
self.md5s[inode] = (mtime, str(size), hash_info.value)
def get(self, path, fs):
"""Gets the hash for the specified path info. Hash will be
retrieved from the state database if available.
Args:
path (str): path info to get the hash for.
Returns:
HashInfo or None: hash for the specified path info or None if it
doesn't exist in the state database.
"""
from .meta import Meta
if not isinstance(fs, LocalFileSystem):
return None, None
try:
mtime, size = get_mtime_and_size(path, fs, self.ignore)
except FileNotFoundError:
return None, None
inode = get_inode(path)
value = self.md5s.get(inode)
if not value or value[0] != mtime or value[1] != str(size):
return None, None
return Meta(size=size), HashInfo("md5", value[2])
def save_link(self, path, fs):
"""Adds the specified path to the list of links created by dvc. This
list is later used on `dvc checkout` to cleanup old links.
Args:
path (str): path info to add to the list of links.
"""
if not isinstance(fs, LocalFileSystem):
return
try:
mtime, _ = get_mtime_and_size(path, fs, self.ignore)
except FileNotFoundError:
return
inode = get_inode(path)
relative_path = relpath(path, self.root_dir)
with self.links as ref:
ref[relative_path] = (inode, mtime)
def get_unused_links(self, used, fs):
"""Removes all saved links except the ones that are used.
Args:
used (list): list of used links that should not be removed.
"""
if not isinstance(fs, LocalFileSystem):
return
unused = []
with self.links as ref:
for relative_path in ref:
path = os.path.join(self.root_dir, relative_path)
if path in used or not fs.exists(path):
continue
inode = get_inode(path)
mtime, _ = get_mtime_and_size(path, fs, self.ignore)
if ref[relative_path] == (inode, mtime):
logger.debug("Removing '%s' as unused link.", path)
unused.append(relative_path)
return unused
def remove_links(self, unused, fs):
if not isinstance(fs, LocalFileSystem):
return
for path in unused:
fs.remove(os.path.join(self.root_dir, path))
with self.links as ref:
for path in unused:
del ref[path]
-44
View File
@@ -1,44 +0,0 @@
import errno
import hashlib
import json
from typing import TYPE_CHECKING, Tuple
if TYPE_CHECKING:
from ._ignore import Ignore
from .fs.base import AnyFSPath, FileSystem
def get_mtime_and_size(
path: "AnyFSPath", fs: "FileSystem", ignore: "Ignore" = None
) -> Tuple[str, int]:
import nanotime
if not fs.isdir(path):
base_stat = fs.info(path)
size = base_stat["size"]
mtime = str(int(nanotime.timestamp(base_stat["mtime"])))
return mtime, size
size = 0
files_mtimes = {}
if ignore:
walk_iterator = ignore.find(fs, path)
else:
walk_iterator = fs.find(path)
for file_path in walk_iterator:
try:
stats = fs.info(file_path)
except OSError as exc:
# NOTE: broken symlink case.
if exc.errno != errno.ENOENT:
raise
continue
size += stats["size"]
files_mtimes[file_path] = stats["mtime"]
# We track file changes and moves, which cannot be detected with simply
# max(mtime(f) for f in non_ignored_files)
hasher = hashlib.md5()
hasher.update(json.dumps(files_mtimes, sort_keys=True).encode("utf-8"))
mtime = hasher.hexdigest()
return mtime, size
+5 -5
View File
@@ -17,9 +17,10 @@ from dvc.exceptions import (
MergeError,
RemoteCacheRequiredError,
)
from dvc.objects.hash_info import HashInfo
from dvc.objects.istextfile import istextfile
from dvc.objects.meta import Meta
from dvc_objects.errors import ObjectFormatError
from dvc_objects.hash_info import HashInfo
from dvc_objects.istextfile import istextfile
from dvc_objects.meta import Meta
from .data import check as ocheck
from .data import load as oload
@@ -35,12 +36,11 @@ from .fs import (
Schemes,
get_cloud_fs,
)
from .objects.errors import ObjectFormatError
from .utils import relpath
from .utils.fs import path_isin
if TYPE_CHECKING:
from .objects.db import ObjectDB
from dvc_objects.db import ObjectDB
logger = logging.getLogger(__name__)
+2 -2
View File
@@ -16,9 +16,9 @@ from dvc.utils.fs import path_isin
if TYPE_CHECKING:
from dvc.fs import FileSystem
from dvc.objects.file import HashFile
from dvc.repo.scm_context import SCMContext
from dvc.scm import Base
from dvc_objects.file import HashFile
logger = logging.getLogger(__name__)
@@ -158,7 +158,6 @@ class Repo:
from dvc.data_cloud import DataCloud
from dvc.fs import GitFileSystem, localfs
from dvc.lock import LockNoop, make_lock
from dvc.objects.state import State, StateNoop
from dvc.odbmgr import ODBManager
from dvc.repo.metrics import Metrics
from dvc.repo.params import Params
@@ -166,6 +165,7 @@ class Repo:
from dvc.repo.stage import StageLoad
from dvc.scm import SCM
from dvc.stage.cache import StageCache
from dvc_objects.state import State, StateNoop
self.url = url
self._fs_conf = {"repo_factory": repo_factory}
+1 -1
View File
@@ -7,7 +7,7 @@ from dvc.fs import Schemes
from . import locked
if TYPE_CHECKING:
from dvc.objects.db.base import ObjectDB
from dvc_objects.db.base import ObjectDB
logger = logging.getLogger(__name__)
+2 -2
View File
@@ -22,12 +22,12 @@ if TYPE_CHECKING:
from dvc.data.tree import Tree
from dvc.dependency import Dependency, ParamsDependency
from dvc.fs import FileSystem
from dvc.objects.db import ObjectDB
from dvc.objects.hash_info import HashInfo
from dvc.output import Output
from dvc.repo.stage import StageLoad
from dvc.stage import Stage
from dvc.types import StrPath, TargetType
from dvc_objects.db import ObjectDB
from dvc_objects.hash_info import HashInfo
ObjectContainer = Dict[Optional["ObjectDB"], Set["HashInfo"]]
+1 -1
View File
@@ -5,7 +5,7 @@ from dvc.repo import locked
from dvc.utils import glob_targets
if TYPE_CHECKING:
from dvc.objects.db.base import ObjectDB
from dvc_objects.db.base import ObjectDB
logger = logging.getLogger(__name__)
+1 -1
View File
@@ -6,7 +6,7 @@ from ..utils import glob_targets
from . import locked
if TYPE_CHECKING:
from dvc.objects.db.base import ObjectDB
from dvc_objects.db.base import ObjectDB
@locked
+1 -1
View File
@@ -3,11 +3,11 @@ from collections.abc import Mapping
from voluptuous import Any, Optional, Required, Schema
from dvc import dependency, output
from dvc.objects.meta import Meta
from dvc.output import CHECKSUMS_SCHEMA, Output
from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
from dvc.parsing.versions import SCHEMA_KWD, lockfile_version_schema
from dvc.stage.params import StageParams
from dvc_objects.meta import Meta
STAGES = "stages"
SINGLE_STAGE_SCHEMA = {
+2 -2
View File
@@ -36,8 +36,8 @@ from .utils import (
if TYPE_CHECKING:
from dvc.dvcfile import DVCFile
from dvc.objects.db.base import ObjectDB
from dvc.objects.hash_info import HashInfo
from dvc_objects.db.base import ObjectDB
from dvc_objects.hash_info import HashInfo
logger = logging.getLogger(__name__)
# Disallow all punctuation characters except hyphen and underscore
+2 -2
View File
@@ -12,7 +12,7 @@ from dvc.exceptions import DvcException
from dvc.utils import dict_sha256, relpath
if TYPE_CHECKING:
from dvc.objects.db import ObjectDB
from dvc_objects.db import ObjectDB
logger = logging.getLogger(__name__)
@@ -23,7 +23,7 @@ class RunCacheNotFoundError(DvcException):
def _get_cache_hash(cache, key=False):
from dvc.objects.meta import Meta
from dvc_objects.meta import Meta
if key:
cache["outs"] = [out["path"] for out in cache.get("outs", [])]
+2 -2
View File
@@ -6,10 +6,10 @@ from itertools import chain
from funcy import cached_property, get_in, lcat, once, project
from dvc import dependency, output
from dvc.objects.hash_info import HashInfo
from dvc.objects.meta import Meta
from dvc.parsing import FOREACH_KWD, JOIN, DataResolver, EntryNotFound
from dvc.parsing.versions import LOCKFILE_VERSION
from dvc_objects.hash_info import HashInfo
from dvc_objects.meta import Meta
from . import PipelineStage, Stage, loads_from
from .exceptions import StageNameUnspecified, StageNotFound
+1 -1
View File
@@ -5,7 +5,7 @@ from typing import TYPE_CHECKING, Union
from funcy import concat, first, lsplit, rpartial, without
from dvc.exceptions import InvalidArgumentError
from dvc.objects.meta import Meta
from dvc_objects.meta import Meta
from .exceptions import (
MissingDataSource,
+1 -1
View File
@@ -25,7 +25,7 @@ exclude = '''
[tool.isort]
profile = "black"
known_first_party = ["dvc", "tests"]
known_first_party = ["dvc", "dvc_objects", "tests"]
line_length = 79
[tool.pytest.ini_options]
+2 -2
View File
@@ -66,7 +66,7 @@ def test_clear_on_download_err(tmp_dir, dvc, index, mocker):
assert list(index.hashes())
mocker.patch("dvc.objects.fs.generic.transfer", side_effect=Exception)
mocker.patch("dvc_objects.fs.generic.transfer", side_effect=Exception)
with pytest.raises(DownloadError):
dvc.pull()
assert not list(index.hashes())
@@ -83,7 +83,7 @@ def test_partial_upload(tmp_dir, dvc, index, mocker):
raise Exception("stop baz")
return original(self, from_file, to_info, name, **kwargs)
mocker.patch("dvc.objects.fs.generic.transfer", unreliable_upload)
mocker.patch("dvc_objects.fs.generic.transfer", unreliable_upload)
with pytest.raises(UploadError):
dvc.push()
assert not list(index.hashes())
+5 -4
View File
@@ -11,6 +11,7 @@ import colorama
import pytest
import dvc as dvc_module
import dvc_objects
from dvc.cli import main
from dvc.dvcfile import DVC_FILE_SUFFIX
from dvc.exceptions import (
@@ -21,8 +22,6 @@ from dvc.exceptions import (
RecursiveAddingWhileUsingFilename,
)
from dvc.fs import LocalFileSystem, system
from dvc.objects.hash import file_md5
from dvc.objects.hash_info import HashInfo
from dvc.odbmgr import ODBManager
from dvc.output import (
OutputAlreadyTrackedError,
@@ -38,6 +37,8 @@ from dvc.testing.test_workspace import TestAdd
from dvc.utils import LARGE_DIR_SIZE, relpath
from dvc.utils.fs import path_isin
from dvc.utils.serialize import YAMLFileCorruptedError, load_yaml
from dvc_objects.hash import file_md5
from dvc_objects.hash_info import HashInfo
from tests.basic_env import TestDvc
from tests.utils import get_gitignore_content
@@ -378,7 +379,7 @@ class TestDoubleAddUnchanged(TestDvc):
def test_should_update_state_entry_for_file_after_add(mocker, dvc, tmp_dir):
file_md5_counter = mocker.spy(dvc_module.objects.hash, "file_md5")
file_md5_counter = mocker.spy(dvc_objects.hash, "file_md5")
tmp_dir.gen("foo", "foo")
ret = main(["config", "cache.type", "copy"])
@@ -409,7 +410,7 @@ def test_should_update_state_entry_for_file_after_add(mocker, dvc, tmp_dir):
def test_should_update_state_entry_for_directory_after_add(
mocker, dvc, tmp_dir
):
file_md5_counter = mocker.spy(dvc_module.objects.hash, "file_md5")
file_md5_counter = mocker.spy(dvc_objects.hash, "file_md5")
tmp_dir.gen({"data/data": "foo", "data/data_sub/sub_data": "foo"})
+4 -4
View File
@@ -5,16 +5,16 @@ import shutil
import pytest
from flaky.flaky_decorator import flaky
import dvc as dvc_module
import dvc_objects
from dvc.cli import main
from dvc.data.db.local import LocalObjectDB
from dvc.external_repo import clean_repos
from dvc.objects.db import ObjectDB
from dvc.stage.exceptions import StageNotFound
from dvc.testing.test_remote import ( # noqa, pylint: disable=unused-import
TestRemote,
)
from dvc.utils.fs import remove
from dvc_objects.db import ObjectDB
def test_cloud_cli(tmp_dir, dvc, remote, mocker):
@@ -149,7 +149,7 @@ def test_warn_on_outdated_stage(tmp_dir, dvc, local_remote, caplog):
def test_hash_recalculation(mocker, dvc, tmp_dir, local_remote):
tmp_dir.gen({"foo": "foo"})
test_file_md5 = mocker.spy(dvc_module.objects.hash, "file_md5")
test_file_md5 = mocker.spy(dvc_objects.hash, "file_md5")
ret = main(["config", "cache.type", "hardlink"])
assert ret == 0
ret = main(["add", "foo"])
@@ -211,7 +211,7 @@ def test_verify_hashes(
remove("dir")
remove(dvc.odb.local.cache_dir)
hash_spy = mocker.spy(dvc_module.objects.hash, "file_md5")
hash_spy = mocker.spy(dvc_objects.hash, "file_md5")
dvc.pull()
assert hash_spy.call_count == 0
+1 -1
View File
@@ -47,7 +47,7 @@ def test_source_change(erepo_dir):
def test_cache_reused(erepo_dir, mocker, local_cloud):
from dvc.objects.fs import generic
from dvc_objects.fs import generic
erepo_dir.add_remote(config=local_cloud.config)
with erepo_dir.chdir():
+1 -1
View File
@@ -6,12 +6,12 @@ import pytest
from dvc.data.stage import IgnoreInCollectedDirError
from dvc.ignore import DvcIgnore, DvcIgnorePatterns
from dvc.objects.utils import get_mtime_and_size
from dvc.output import OutputIsIgnoredError
from dvc.pathspec_math import PatternInfo, merge_patterns
from dvc.repo import Repo
from dvc.testing.tmp_dir import TmpDir
from dvc.types import List
from dvc_objects.utils import get_mtime_and_size
def _to_pattern_info_list(str_list: List):
+1 -1
View File
@@ -308,7 +308,7 @@ def test_download_error_pulling_imported_stage(tmp_dir, dvc, erepo_dir):
remove(dst_cache)
with patch(
"dvc.objects.fs.generic.transfer", side_effect=Exception
"dvc_objects.fs.generic.transfer", side_effect=Exception
), pytest.raises(DownloadError):
dvc.pull(["foo_imported.dvc"])
+1 -1
View File
@@ -6,7 +6,7 @@ import pytest
from git import GitCommandError
from dvc.exceptions import DvcException
from dvc.objects.hash import file_md5
from dvc_objects.hash import file_md5
from tests.func.parsing.test_errors import escape_ansi
+1 -1
View File
@@ -6,11 +6,11 @@ from operator import itemgetter
import pytest
from dvc.dvcfile import PIPELINE_LOCK, Lockfile
from dvc.objects.hash_info import HashInfo
from dvc.stage.utils import split_params_deps
from dvc.utils.fs import remove
from dvc.utils.serialize import dumps_yaml, parse_yaml_for_update
from dvc.utils.strictyaml import YAMLValidationError, make_relpath
from dvc_objects.hash_info import HashInfo
from tests.func.test_run_multistage import supported_params
FS_STRUCTURE = {
+2 -2
View File
@@ -5,10 +5,10 @@ import configobj
import pytest
from dvc.cli import main
from dvc.objects.errors import ObjectFormatError
from dvc.objects.hash_info import HashInfo
from dvc.odbmgr import ODBManager
from dvc.utils import relpath
from dvc_objects.errors import ObjectFormatError
from dvc_objects.hash_info import HashInfo
def test_cache(tmp_dir, dvc):
+3 -3
View File
@@ -169,7 +169,7 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc):
def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote):
from dvc.objects.fs import generic
from dvc_objects.fs import generic
foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0]
bar = tmp_dir.dvc_gen({"bar": "bar content"})[0].outs[0]
@@ -214,7 +214,7 @@ def test_raise_on_too_many_open_files(
tmp_dir.dvc_gen({"file": "file content"})
mocker.patch(
"dvc.objects.fs.generic.transfer",
"dvc_objects.fs.generic.transfer",
side_effect=OSError(errno.EMFILE, "Too many open files"),
)
@@ -263,7 +263,7 @@ def test_external_dir_resource_on_no_cache(tmp_dir, dvc, tmp_path_factory):
def test_push_order(tmp_dir, dvc, tmp_path_factory, mocker, local_remote):
from dvc.objects.fs import generic
from dvc_objects.fs import generic
foo = tmp_dir.dvc_gen({"foo": {"bar": "bar content"}})[0].outs[0]
tmp_dir.dvc_gen({"baz": "baz content"})
+1 -1
View File
@@ -4,9 +4,9 @@ import pytest
from dvc.cli import main
from dvc.fs import system
from dvc.objects.errors import ObjectDBError
from dvc.stage.exceptions import StageFileDoesNotExistError
from dvc.utils.fs import remove
from dvc_objects.errors import ObjectDBError
from tests.utils import get_gitignore_content
+1 -1
View File
@@ -258,7 +258,7 @@ def get_index(dvc, rev):
@pytest.mark.parametrize("rev", ["workspace", "HEAD"])
def test_used_objs(tmp_dir, scm, dvc, run_copy, rev):
from dvc.objects.hash_info import HashInfo
from dvc_objects.hash_info import HashInfo
dvc.config["core"]["autostage"] = True
tmp_dir.dvc_gen({"dir": {"subdir": {"file": "file"}}, "foo": "foo"})
+2 -2
View File
@@ -15,13 +15,13 @@ from dvc.exceptions import (
StagePathAsOutputError,
)
from dvc.fs import LocalFileSystem, system
from dvc.objects.hash import file_md5
from dvc.output import Output
from dvc.stage import Stage
from dvc.stage.exceptions import StageFileDoesNotExistError
from dvc.utils import relpath
from dvc.utils.fs import remove
from dvc.utils.serialize import dump_yaml, load_yaml
from dvc_objects.hash import file_md5
from tests.basic_env import TestDvc
@@ -817,7 +817,7 @@ class TestReproAllPipelines(SingleStageRun, TestDvc):
),
]
from dvc.objects.state import StateNoop
from dvc_objects.state import StateNoop
self.dvc.state = StateNoop()
+1 -1
View File
@@ -20,7 +20,6 @@ from dvc.exceptions import (
StagePathAsOutputError,
)
from dvc.fs import system
from dvc.objects.hash import file_md5
from dvc.output import Output, OutputIsStageFileError
from dvc.repo import Repo as DvcRepo
from dvc.stage import Stage
@@ -32,6 +31,7 @@ from dvc.stage.exceptions import (
StagePathOutsideError,
)
from dvc.utils.serialize import load_yaml
from dvc_objects.hash import file_md5
from tests.basic_env import TestDvc, TestDvcGit
+3 -3
View File
@@ -1,10 +1,10 @@
import os
import re
from dvc.objects.hash import file_md5
from dvc.objects.hash_info import HashInfo
from dvc.objects.state import State
from dvc.repo import Repo
from dvc_objects.hash import file_md5
from dvc_objects.hash_info import HashInfo
from dvc_objects.state import State
def test_state(tmp_dir, dvc):
+2 -2
View File
@@ -5,8 +5,8 @@ from funcy import raiser
from dvc.cli import main
from dvc.data.stage import IgnoreInCollectedDirError
from dvc.objects.cache import DiskError
from dvc.objects.fs.base import FileSystem, RemoteMissingDepsError
from dvc_objects.cache import DiskError
from dvc_objects.fs.base import FileSystem, RemoteMissingDepsError
def test_state_pickle_errors_are_correctly_raised(tmp_dir, caplog, mocker):
+1 -1
View File
@@ -5,7 +5,7 @@ import pytest
from dvc.data.db.local import LocalObjectDB
from dvc.fs import LocalFileSystem
from dvc.objects.hash_info import HashInfo
from dvc_objects.hash_info import HashInfo
def test_status_download_optimization(mocker, dvc):
+2 -2
View File
@@ -3,8 +3,8 @@ from operator import itemgetter
import pytest
from dvc.data.tree import Tree, _merge
from dvc.objects.hash_info import HashInfo
from dvc.objects.meta import Meta
from dvc_objects.hash_info import HashInfo
from dvc_objects.meta import Meta
@pytest.mark.parametrize(
+3 -4
View File
@@ -3,11 +3,12 @@ import shutil
import pytest
import dvc_objects
from dvc.config import NoRemoteError
from dvc.data.stage import stage
from dvc.fs.data import DataFileSystem
from dvc.objects.hash_info import HashInfo
from dvc.utils.fs import remove
from dvc_objects.hash_info import HashInfo
@pytest.mark.parametrize(
@@ -226,13 +227,11 @@ def test_get_hash_file(tmp_dir, dvc):
def test_get_hash_dir(tmp_dir, dvc, mocker):
import dvc as dvc_module
tmp_dir.dvc_gen(
{"dir": {"foo": "foo", "bar": "bar", "subdir": {"data": "data"}}}
)
fs = DataFileSystem(repo=dvc)
hash_file_spy = mocker.spy(dvc_module.objects.hash, "hash_file")
hash_file_spy = mocker.spy(dvc_objects.hash, "hash_file")
assert fs.info("dir")["md5"] == "8761c4e9acad696bee718615e23e22db.dir"
assert not hash_file_spy.called
+3 -3
View File
@@ -7,7 +7,7 @@ import pytest
from dvc.data.stage import stage
from dvc.fs.dvc import DvcFileSystem
from dvc.objects.hash_info import HashInfo
from dvc_objects.hash_info import HashInfo
from tests.utils import clean_staging
@@ -576,8 +576,8 @@ def test_get_hash_mixed_dir(tmp_dir, scm, dvc):
def test_get_hash_dirty_file(tmp_dir, dvc):
from dvc.data import check
from dvc.objects.errors import ObjectFormatError
from dvc.objects.hash import hash_file
from dvc_objects.errors import ObjectFormatError
from dvc_objects.hash import hash_file
tmp_dir.dvc_gen("file", "file")
file_hash_info = HashInfo("md5", "8c7dd922ad47494fc02c388e12c00eac")
+1 -1
View File
@@ -45,7 +45,7 @@ def test_init():
def test_ssh_ask_password(mocker):
mocker.patch(
"dvc.objects.fs.implementations.ssh.ask_password", return_value="fish"
"dvc_objects.fs.implementations.ssh.ask_password", return_value="fish"
)
fs = SSHFileSystem(user="test", host="2.2.2.2", ask_password=True)
assert fs.fs_args["password"] == fs.fs_args["passphrase"] == "fish"
+1 -1
View File
@@ -3,7 +3,7 @@ from typing import Any
import pytest
from dvc.objects.cache import Cache, DiskError
from dvc_objects.cache import Cache, DiskError
def set_value(cache: Cache, key: str, value: Any) -> Any:
+2 -2
View File
@@ -1,5 +1,5 @@
from dvc.objects.fs import LocalFileSystem
from dvc.objects.hash import file_md5
from dvc_objects.fs import LocalFileSystem
from dvc_objects.hash import file_md5
def test_file_md5(tmp_path):
+3 -3
View File
@@ -1,8 +1,8 @@
import pytest
from dvc.objects.fs import LocalFileSystem
from dvc.objects.hash import HashStreamFile, file_md5
from dvc.objects.istextfile import DEFAULT_CHUNK_SIZE, istextfile
from dvc_objects.fs import LocalFileSystem
from dvc_objects.hash import HashStreamFile, file_md5
from dvc_objects.istextfile import DEFAULT_CHUNK_SIZE, istextfile
def test_hashed_stream_reader(tmp_path):
+2 -2
View File
@@ -1,11 +1,11 @@
import os
from unittest.mock import patch
from dvc.objects.hash_info import HashInfo
from dvc.objects.meta import Meta
from dvc.output import Output
from dvc.stage import Stage
from dvc.utils import relpath
from dvc_objects.hash_info import HashInfo
from dvc_objects.meta import Meta
from tests.basic_env import TestDvc
+1 -1
View File
@@ -3,7 +3,7 @@ import posixpath
from unittest import mock
from dvc.fs import FileSystem
from dvc.objects.db import ObjectDB
from dvc_objects.db import ObjectDB
class _CallableOrNone:
+1 -1
View File
@@ -65,7 +65,7 @@ def test_token():
)
@patch("dvc.objects.fs.implementations.webdav.ask_password")
@patch("dvc_objects.fs.implementations.webdav.ask_password")
def test_ask_password(ask_password_mocked):
ask_password_mocked.return_value = "pass"
host = "host"
+1 -1
View File
@@ -4,8 +4,8 @@ import shutil
import pytest
from dvc.exceptions import OutputDuplicationError
from dvc.objects.hash_info import HashInfo
from dvc.repo import NotDvcRepoError, Repo, locked
from dvc_objects.hash_info import HashInfo
def test_is_dvc_internal(dvc):
@@ -5,10 +5,10 @@ from itertools import chain
import pytest
from dvc.dvcfile import PIPELINE_FILE, Dvcfile
from dvc.objects.hash_info import HashInfo
from dvc.stage import PipelineStage, create_stage
from dvc.stage.loader import StageLoader
from dvc.stage.serialize import split_params_deps
from dvc_objects.hash_info import HashInfo
@pytest.fixture

Some files were not shown because too many files have changed in this diff Show More