dvc: bump to 2.7.3

Bumps [adlfs](https://github.com/dask/adlfs) from 2021.8.2 to 2021.9.1.
- [Release notes](https://github.com/dask/adlfs/releases)
- [Changelog](https://github.com/dask/adlfs/blob/master/CHANGELOG.md)
- [Commits](https://github.com/dask/adlfs/compare/2021.08.2...2021.09.1)

---
updated-dependencies:
- dependency-name: adlfs
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <[email protected]>

Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/packages.yaml

@@ -5,6 +5,7 @@ on:
     - master
   release:
     types: [released, prereleased]
+  workflow_dispatch:
 jobs:
   deb:
     runs-on: ubuntu-18.04

.github/workflows/tests.yaml

@@ -13,6 +13,7 @@ env:
 
 jobs:
   lint:
+    timeout-minutes: 10
     runs-on: ubuntu-latest
     steps:
     - uses: actions/[email protected]
@@ -32,6 +33,7 @@ jobs:
       run: python setup.py checkdocs
     - uses: pre-commit/[email protected]
   tests:
+    timeout-minutes: 45
     runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
@@ -97,6 +99,7 @@ jobs:
       run: |
         echo "extra_test_args=--disable-gdrive" >> $GITHUB_ENV
     - name: run tests
+      timeout-minutes: 30
       env:
         GDRIVE_CREDENTIALS_DATA: ${{ secrets.GDRIVE_CREDENTIALS_DATA }}
         AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}

MANIFEST.in

@@ -1,3 +1,3 @@
 include fastentrypoints.py
 include LICENSE
-include test_requirements.txt
+recursive-include requirements *.txt

README.rst

@@ -222,7 +222,9 @@ Contributing
 |Maintainability| |Donate|
 
 Contributions are welcome! Please see our `Contributing Guide <https://dvc.org/doc/user-guide/contributing/core>`_ for more
-details.
+details. Thanks to all our contributors!
+
+|Contribs|
 
 Mailing List
 ============
@@ -299,3 +301,7 @@ Barrak, A., Eghan, E.E. and Adams, B. `On the Co-evolution of ML Pipelines and S
 .. |Flowchart| image:: https://dvc.org/img/flow.gif
    :target: https://dvc.org/img/flow.gif
    :alt: how_dvc_works
+
+.. |Contribs| image:: https://contrib.rocks/image?repo=iterative/dvc
+   :target: https://github.com/iterative/dvc/graphs/contributors
+   :alt: Contributors

dvc/command/experiments.py

@@ -141,6 +141,8 @@ def _collect_rows(
     precision=DEFAULT_PRECISION,
     sort_by=None,
     sort_order=None,
+    fill_value=FILL_VALUE,
+    iso=False,
 ):
     from dvc.scm.git import Git
 
@@ -161,8 +163,8 @@ def _collect_rows(
         elif exp.get("queued"):
             state = "Queued"
         else:
-            state = FILL_VALUE
-        executor = exp.get("executor", FILL_VALUE)
+            state = fill_value
+        executor = exp.get("executor", fill_value)
         is_baseline = rev == "baseline"
 
         if is_baseline:
@@ -202,12 +204,12 @@ def _collect_rows(
             exp_name,
             name_rev,
             typ,
-            _format_time(exp.get("timestamp")),
+            _format_time(exp.get("timestamp"), fill_value, iso),
             parent,
             state,
             executor,
         ]
-        fill_value = FILL_VALUE_ERRORED if results.get("error") else FILL_VALUE
+        fill_value = FILL_VALUE_ERRORED if results.get("error") else fill_value
         _extend_row(
             row,
             metric_names,
@@ -274,14 +276,18 @@ def _sort_exp(experiments, sort_path, sort_name, typ, reverse):
     return ret
 
 
-def _format_time(timestamp):
-    if timestamp is None:
-        return FILL_VALUE
-    if timestamp.date() == date.today():
+def _format_time(datetime_obj, fill_value=FILL_VALUE, iso=False):
+    if datetime_obj is None:
+        return fill_value
+
+    if iso:
+        return datetime_obj.isoformat()
+
+    if datetime_obj.date() == date.today():
         fmt = "%I:%M %p"
     else:
         fmt = "%b %d, %Y"
-    return timestamp.strftime(fmt)
+    return datetime_obj.strftime(fmt)
 
 
 def _extend_row(row, names, items, precision, fill_value=FILL_VALUE):
@@ -320,6 +326,7 @@ def _parse_filter_list(param_list):
 
 def experiments_table(
     all_experiments,
+    headers,
     metric_headers,
     metric_names,
     param_headers,
@@ -327,22 +334,15 @@ def experiments_table(
     sort_by=None,
     sort_order=None,
     precision=DEFAULT_PRECISION,
+    fill_value=FILL_VALUE,
+    iso=False,
 ) -> "TabularData":
     from funcy import lconcat
 
     from dvc.compare import TabularData
 
-    headers = [
-        "Experiment",
-        "rev",
-        "typ",
-        "Created",
-        "parent",
-        "State",
-        "Executor",
-    ]
     td = TabularData(
-        lconcat(headers, metric_headers, param_headers), fill_value=FILL_VALUE
+        lconcat(headers, metric_headers, param_headers), fill_value=fill_value
     )
     for base_rev, experiments in all_experiments.items():
         rows = _collect_rows(
@@ -353,6 +353,8 @@ def experiments_table(
             sort_by=sort_by,
             sort_order=sort_order,
             precision=precision,
+            fill_value=fill_value,
+            iso=iso,
         )
         td.extend(rows)
 
@@ -385,8 +387,10 @@ def baseline_styler(typ):
 
 
 def show_experiments(
-    all_experiments, pager=True, no_timestamp=False, **kwargs
+    all_experiments, pager=True, no_timestamp=False, show_csv=False, **kwargs
 ):
+    from funcy.seqs import flatten as flatten_list
+
     include_metrics = _parse_filter_list(kwargs.pop("include_metrics", []))
     exclude_metrics = _parse_filter_list(kwargs.pop("exclude_metrics", []))
     include_params = _parse_filter_list(kwargs.pop("include_params", []))
@@ -399,11 +403,26 @@ def show_experiments(
         include_params=include_params,
         exclude_params=exclude_params,
     )
-    metric_headers = _normalize_headers(metric_names)
-    param_headers = _normalize_headers(param_names)
+
+    headers = [
+        "Experiment",
+        "rev",
+        "typ",
+        "Created",
+        "parent",
+        "State",
+        "Executor",
+    ]
+
+    names = {**metric_names, **param_names}
+    counter = Counter(flatten_list([list(a.keys()) for a in names.values()]))
+    counter.update(headers)
+    metric_headers = _normalize_headers(metric_names, counter)
+    param_headers = _normalize_headers(param_names, counter)
 
     td = experiments_table(
         all_experiments,
+        headers,
         metric_headers,
         metric_names,
         param_headers,
@@ -411,6 +430,8 @@ def show_experiments(
         kwargs.get("sort_by"),
         kwargs.get("sort_order"),
         kwargs.get("precision"),
+        kwargs.get("fill_value"),
+        kwargs.get("iso"),
     )
 
     if no_timestamp:
@@ -422,9 +443,12 @@ def show_experiments(
 
     row_styles = lmap(baseline_styler, td.column("typ"))
 
-    merge_headers = ["Experiment", "rev", "typ", "parent"]
-    td.column("Experiment")[:] = map(prepare_exp_id, td.as_dict(merge_headers))
-    td.drop(*merge_headers[1:])
+    if not show_csv:
+        merge_headers = ["Experiment", "rev", "typ", "parent"]
+        td.column("Experiment")[:] = map(
+            prepare_exp_id, td.as_dict(merge_headers)
+        )
+        td.drop(*merge_headers[1:])
 
     headers = {"metrics": metric_headers, "params": param_headers}
     styles = {
@@ -453,13 +477,11 @@ def show_experiments(
         rich_table=True,
         header_styles=styles,
         row_styles=row_styles,
+        show_csv=show_csv,
     )
 
 
-def _normalize_headers(names):
-    count = Counter(
-        name for path in names for name in names[path] for path in names
-    )
+def _normalize_headers(names, count):
     return [
         name if count[name] == 1 else f"{path}:{name}"
         for path in names
@@ -493,6 +515,14 @@ class CmdExperimentsShow(CmdBase):
 
             ui.write(json.dumps(all_experiments, default=_format_json))
         else:
+            precision = (
+                self.args.precision or None
+                if self.args.show_csv
+                else DEFAULT_PRECISION
+            )
+            fill_value = "" if self.args.show_csv else FILL_VALUE
+            iso = True if self.args.show_csv else False
+
             show_experiments(
                 all_experiments,
                 include_metrics=self.args.include_metrics,
@@ -502,8 +532,11 @@ class CmdExperimentsShow(CmdBase):
                 no_timestamp=self.args.no_timestamp,
                 sort_by=self.args.sort_by,
                 sort_order=self.args.sort_order,
-                precision=self.args.precision or DEFAULT_PRECISION,
+                precision=precision,
+                fill_value=fill_value,
+                iso=iso,
                 pager=not self.args.no_pager,
+                show_csv=self.args.show_csv,
             )
         return 0
 
@@ -883,6 +916,12 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Print output in JSON format instead of a human-readable table.",
     )
+    experiments_show_parser.add_argument(
+        "--show-csv",
+        action="store_true",
+        default=False,
+        help="Print output in csv format instead of a human-readable table.",
+    )
     experiments_show_parser.add_argument(
         "--precision",
         type=int,
@@ -1112,9 +1151,9 @@ def add_parser(subparsers, parent_parser):
         nargs="?",
         default=None,
         help=(
-            "Optional Git remote name or Git URL. If provided, experiments "
-            "from the specified Git repository will be listed instead of "
-            "local experiments."
+            "Optional Git remote name or Git URL. "
+            "If provided, experiments from the specified Git repository "
+            " will be listed instead of local ones."
         ),
         metavar="[<git_remote>]",
     )

dvc/compare.py

@@ -168,7 +168,10 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
     def render(self, **kwargs: Any):
         from dvc.ui import ui
 
-        ui.table(self, headers=self.keys(), **kwargs)
+        if kwargs.pop("show_csv", False):
+            ui.write(self.to_csv(), end="")
+        else:
+            ui.table(self, headers=self.keys(), **kwargs)
 
     def as_dict(
         self, cols: Iterable[str] = None

dvc/config.py

@@ -263,6 +263,7 @@ class Config(dict):
                     "key_path": func,
                 }
             },
+            "machine": {str: {"startup_script": func}},
         }
         return Schema(dirs_schema, extra=ALLOW_EXTRA)(conf)
 

dvc/config_schema.py

@@ -229,6 +229,8 @@ SCHEMA = {
     "machine": {
         str: {
             "cloud": All(Lower, Choices("aws", "azure")),
+            "instance_type": Lower,
+            "startup_script": str,
         },
     },
     # section for experimental features

dvc/fs/base.py

@@ -217,44 +217,50 @@ class BaseFileSystem:
             return False
         return hash_.endswith(cls.CHECKSUM_DIR_SUFFIX)
 
-    def upload(self, from_info, to_info, name=None, no_progress_bar=False):
-        if not hasattr(self, "_upload"):
-            raise RemoteActionNotImplemented("upload", self.scheme)
+    def upload(
+        self,
+        from_info,
+        to_info,
+        total=None,
+        desc=None,
+        no_progress_bar=False,
+        **pbar_args,
+    ):
+        is_file_obj = hasattr(from_info, "read")
+        method = "upload_fobj" if is_file_obj else "_upload"
+        if not hasattr(self, method):
+            raise RemoteActionNotImplemented(method, self.scheme)
 
         if to_info.scheme != self.scheme:
             raise NotImplementedError
 
+        logger.debug("Uploading '%s' to '%s'", from_info, to_info)
+        if is_file_obj:
+            with Tqdm.wrapattr(
+                from_info,
+                "read",
+                disable=no_progress_bar,
+                bytes=True,
+                total=total,
+                desc=desc,
+                **pbar_args,
+            ) as wrapped:
+                # `size` is used to provide hints to the WebdavFileSystem
+                # for legacy servers.
+                # pylint: disable=no-member
+                return self.upload_fobj(wrapped, to_info, size=total)
+
         if from_info.scheme != "local":
             raise NotImplementedError
 
-        logger.debug("Uploading '%s' to '%s'", from_info, to_info)
-
-        name = name or from_info.name
-
-        self._upload(  # noqa, pylint: disable=no-member
+        name = desc or from_info.name
+        return self._upload(  # noqa, pylint: disable=no-member
             from_info.fspath,
             to_info,
             name=name,
             no_progress_bar=no_progress_bar,
-        )
-
-    def upload_fobj(
-        self, fobj, to_info, no_progress_bar=False, size=None, **pbar_args
-    ):
-        if not hasattr(self, "_upload_fobj"):
-            raise RemoteActionNotImplemented("upload_fobj", self.scheme)
-
-        with Tqdm.wrapattr(
-            fobj,
-            "read",
-            disable=no_progress_bar,
-            bytes=True,
-            total=size,
             **pbar_args,
-        ) as wrapped:
-            self._upload_fobj(  # pylint: disable=no-member
-                wrapped, to_info, size=size
-            )
+        )
 
     def download(
         self,

dvc/fs/fsspec_wrapper.py

@@ -130,7 +130,7 @@ class FSSpecWrapper(BaseFileSystem):
             self._with_bucket(path_info), exist_ok=kwargs.pop("exist_ok", True)
         )
 
-    def _upload_fobj(self, fobj, to_info, size=None):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         self.makedirs(to_info.parent)
         with self.open(to_info, "wb") as fdest:
             shutil.copyfileobj(fobj, fdest, length=fdest.blocksize)
@@ -141,13 +141,15 @@ class FSSpecWrapper(BaseFileSystem):
         self.makedirs(to_info.parent)
         size = os.path.getsize(from_file)
         with open(from_file, "rb") as fobj:
-            self.upload_fobj(
+            with Tqdm.wrapattr(
                 fobj,
-                to_info,
-                size=size,
+                "read",
+                disable=no_progress_bar,
+                bytes=True,
+                total=size,
                 desc=name,
-                no_progress_bar=no_progress_bar,
-            )
+            ) as wrapped:
+                self.upload_fobj(wrapped, to_info, size=size)
         self.fs.invalidate_cache(self._with_bucket(to_info.parent))
 
     def _download(
@@ -233,6 +235,27 @@ class ObjectFSWrapper(FSSpecWrapper):
         yield from self._strip_buckets(files, detail=detail)
 
 
+# pylint: disable=arguments-differ
+class NoDirectoriesMixin:
+    def isdir(self, *args, **kwargs):
+        return False
+
+    def isfile(self, *args, **kwargs):
+        return True
+
+    def find(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def walk(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def walk_files(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def ls(self, *args, **kwargs):
+        raise NotImplementedError
+
+
 _LOCAL_FS = LocalFileSystem()
 
 

dvc/fs/gdrive.py

@@ -298,7 +298,7 @@ class GDriveFileSystem(
 
         return super()._with_bucket(path)
 
-    def _upload_fobj(self, fobj, to_info, size: int = None):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         rpath = self._with_bucket(to_info)
         self.makedirs(os.path.dirname(rpath))
-        return self.fs.upload_fobj(fobj, rpath, size=size)
+        return self.fs.upload_fobj(fobj, rpath, **kwargs)

dvc/fs/hdfs.py

@@ -244,7 +244,7 @@ class HDFSFileSystem(BaseFileSystem):
             size=self.getsize(path_info),
         )
 
-    def _upload_fobj(self, fobj, to_info, **kwargs):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         with self.hdfs(to_info) as hdfs:
             with hdfs.open_output_stream(to_info.path) as fdest:
                 shutil.copyfileobj(fobj, fdest, self.BLOCK_SIZE)

dvc/fs/http.py

@@ -1,20 +1,12 @@
-import logging
-import os.path
 import threading
-from typing import Optional
-from urllib.parse import urlparse
 
-from funcy import cached_property, memoize, wrap_prop, wrap_with
+from funcy import cached_property, memoize, wrap_with
 
 from dvc import prompt
-from dvc.exceptions import DvcException, HTTPError
 from dvc.path_info import HTTPURLInfo
-from dvc.progress import Tqdm
 from dvc.scheme import Schemes
 
-from .base import BaseFileSystem
-
-logger = logging.getLogger(__name__)
+from .fsspec_wrapper import CallbackMixin, FSSpecWrapper, NoDirectoriesMixin
 
 
 @wrap_with(threading.Lock())
@@ -26,179 +18,111 @@ def ask_password(host, user):
     )
 
 
-class HTTPFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
+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(CallbackMixin, NoDirectoriesMixin, FSSpecWrapper):
     scheme = Schemes.HTTP
     PATH_CLS = HTTPURLInfo
-    PARAM_CHECKSUM = "etag"
+    PARAM_CHECKSUM = "checksum"
+    REQUIRES = {"aiohttp": "aiohttp", "aiohttp-retry": "aiohttp_retry"}
     CAN_TRAVERSE = False
-    REQUIRES = {"requests": "requests"}
 
     SESSION_RETRIES = 5
     SESSION_BACKOFF_FACTOR = 0.1
     REQUEST_TIMEOUT = 60
-    CHUNK_SIZE = 2 ** 16
 
-    def __init__(self, **config):
-        super().__init__(**config)
+    def _prepare_credentials(self, **config):
+        import aiohttp
+        from fsspec.asyn import fsspec_loop
 
-        self.user = config.get("user", None)
+        from dvc.config import ConfigError
 
-        self.auth = config.get("auth", None)
-        self.custom_auth_header = config.get("custom_auth_header", None)
-        self.password = config.get("password", None)
-        self.ask_password = config.get("ask_password", False)
-        self.headers = {}
-        self.ssl_verify = config.get("ssl_verify", True)
-        self.method = config.get("method", "POST")
+        credentials = {}
+        client_kwargs = credentials.setdefault("client_kwargs", {})
 
-    def _auth_method(self, url):
-        from requests.auth import HTTPBasicAuth, HTTPDigestAuth
+        if config.get("auth"):
+            user = config.get("user")
+            password = config.get("password")
+            custom_auth_header = config.get("custom_auth_header")
 
-        if self.auth:
-            if self.ask_password and self.password is None:
-                self.password = ask_password(urlparse(url).hostname, self.user)
-            if self.auth == "basic":
-                return HTTPBasicAuth(self.user, self.password)
-            if self.auth == "digest":
-                return HTTPDigestAuth(self.user, self.password)
-            if self.auth == "custom" and self.custom_auth_header:
-                self.headers.update({self.custom_auth_header: self.password})
-        return None
+            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."
+                )
+
+        if "ssl_verify" in config:
+            with fsspec_loop():
+                client_kwargs["connector"] = aiohttp.TCPConnector(
+                    ssl=make_context(config["ssl_verify"])
+                )
+
+        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,
+        )
+
+        # 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=None,
+        )
+
+        return RetryClient(**kwargs)
 
-    @wrap_prop(threading.Lock())
     @cached_property
-    def _session(self):
-        import requests
-        from requests.adapters import HTTPAdapter
-        from urllib3.util.retry import Retry
-
-        session = requests.Session()
-
-        session.verify = self.ssl_verify
-
-        retries = Retry(
-            total=self.SESSION_RETRIES,
-            backoff_factor=self.SESSION_BACKOFF_FACTOR,
+    def fs(self):
+        from fsspec.implementations.http import (
+            HTTPFileSystem as _HTTPFileSystem,
         )
 
-        session.mount("http://", HTTPAdapter(max_retries=retries))
-        session.mount("https://", HTTPAdapter(max_retries=retries))
+        return _HTTPFileSystem(**self.fs_args)
 
-        return session
-
-    def request(self, method, url, **kwargs):
-        import requests
-
-        kwargs.setdefault("allow_redirects", True)
-        kwargs.setdefault("timeout", self.REQUEST_TIMEOUT)
-
-        try:
-            res = self._session.request(
-                method,
-                url,
-                auth=self._auth_method(url),
-                headers=self.headers,
-                **kwargs,
-            )
-
-            redirect_no_location = (
-                kwargs["allow_redirects"]
-                and res.status_code in (301, 302)
-                and "location" not in res.headers
-            )
-
-            if redirect_no_location:
-                # AWS s3 doesn't like to add a location header to its redirects
-                # from https://s3.amazonaws.com/<bucket name>/* type URLs.
-                # This should be treated as an error
-                raise requests.exceptions.RequestException
-
-            return res
-
-        except requests.exceptions.RequestException:
-            raise DvcException(f"could not perform a {method} request")
-
-    def _head(self, url):
-        response = self.request("HEAD", url)
-        if response.ok:
-            return response
-
-        # Sometimes servers are configured to forbid HEAD requests
-        # Context: https://github.com/iterative/dvc/issues/4131
-        with self.request("GET", url, stream=True) as r:
-            if r.ok:
-                return r
-
-        return response
-
-    def exists(self, path_info) -> bool:
-        res = self._head(path_info.url)
-        if res.status_code == 404:
-            return False
-        if bool(res):
-            return True
-        raise HTTPError(res.status_code, res.reason)
-
-    def info(self, path_info):
-        resp = self._head(path_info.url)
-        etag = resp.headers.get("ETag") or resp.headers.get("Content-MD5")
-        size = self._content_length(resp)
-        return {"etag": etag, "size": size, "type": "file"}
-
-    def _upload_fobj(self, fobj, to_info, **kwargs):
-        def chunks(fobj):
-            while True:
-                chunk = fobj.read(self.CHUNK_SIZE)
-                if not chunk:
-                    break
-                yield chunk
-
-        response = self.request(self.method, to_info.url, data=chunks(fobj))
-        if response.status_code not in (200, 201):
-            raise HTTPError(response.status_code, response.reason)
-
-    def _download(self, from_info, to_file, name=None, no_progress_bar=False):
-        response = self.request("GET", from_info.url, stream=True)
-        if response.status_code != 200:
-            raise HTTPError(response.status_code, response.reason)
-        with open(to_file, "wb") as fd:
-            with Tqdm.wrapattr(
-                fd,
-                "write",
-                total=None
-                if no_progress_bar
-                else self._content_length(response),
-                leave=False,
-                desc=from_info.url if name is None else name,
-                disable=no_progress_bar,
-            ) as fd_wrapped:
-                for chunk in response.iter_content(chunk_size=self.CHUNK_SIZE):
-                    fd_wrapped.write(chunk)
-
-    def _upload(
-        self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
-    ):
-        with open(from_file, "rb") as fobj:
-            self.upload_fobj(
-                fobj,
-                to_info,
-                size=None if no_progress_bar else os.path.getsize(from_file),
-                no_progress_bar=no_progress_bar,
-                desc=name or to_info.url,
-            )
-
-    def open(self, path_info, mode: str = "r", encoding: str = None, **kwargs):
-        from dvc.utils.http import open_url
-
-        return open_url(
-            path_info.url,
-            mode=mode,
-            encoding=encoding,
-            auth=self._auth_method(path_info),
-            **kwargs,
-        )
-
-    @staticmethod
-    def _content_length(response) -> Optional[int]:
-        res = response.headers.get("Content-Length")
-        return int(res) if res else None
+    def _entry_hook(self, entry):
+        entry["checksum"] = entry.get("ETag") or entry.get("Content-MD5")
+        return entry

dvc/fs/local.py

@@ -108,7 +108,7 @@ class LocalFileSystem(BaseFileSystem):
             self.remove(tmp_info)
             raise
 
-    def _upload_fobj(self, fobj, to_info, **kwargs):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         self.makedirs(to_info.parent)
         tmp_info = to_info.parent / tmp_fname("")
         try:

dvc/fs/ssh.py

@@ -67,6 +67,7 @@ class SSHFileSystem(CallbackMixin, FSSpecWrapper):
         )
 
         login_info["password"] = config.get("password")
+        login_info["passphrase"] = config.get("password")
 
         raw_keys = []
         if config.get("keyfile"):
@@ -119,9 +120,9 @@ class SSHFileSystem(CallbackMixin, FSSpecWrapper):
     # Ensure that if an interrupt happens during the transfer, we don't
     # pollute the cache.
 
-    def _upload_fobj(self, fobj, to_info, *args, **kwargs):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         with as_atomic(self, to_info) as tmp_file:
-            super()._upload_fobj(fobj, tmp_file, *args, **kwargs)
+            super().upload_fobj(fobj, tmp_file, **kwargs)
 
     def _upload(self, from_file, to_info, *args, **kwargs):
         with as_atomic(self, to_info) as tmp_file:

dvc/fs/utils.py

@@ -33,7 +33,8 @@ def transfer(
             return from_fs.download_file(from_info, to_info)
 
         with from_fs.open(from_info, mode="rb") as fobj:
-            return to_fs.upload_fobj(fobj, to_info)
+            size = from_fs.getsize(from_info)
+            return to_fs.upload(fobj, to_info, total=size)
     except OSError as exc:
         # If the target file already exists, we are going to simply
         # ignore the exception (#4992).

dvc/fs/webdav.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import threading
 from functools import lru_cache
 
@@ -71,15 +70,13 @@ class WebDAVFileSystem(FSSpecWrapper):  # pylint:disable=abstract-method
 
         return WebdavFileSystem(**self.fs_args)
 
-    def _upload_fobj(self, fobj, to_info, size: int = None):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         rpath = self.translate_path_info(to_info)
-        self.makedirs(os.path.dirname(rpath))
+        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.client.upload_fileobj(
-            fobj, rpath, overwrite=True, size=size
-        )
+        return self.fs.upload_fileobj(fobj, rpath, overwrite=True, size=size)
 
     @lru_cache(512)
     def translate_path_info(self, path):

dvc/fs/webhdfs.py

@@ -141,7 +141,7 @@ class WebHDFSFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         self.hdfs_client.makedirs(to_info.parent.path)
         self.hdfs_client.rename(from_info.path, to_info.path)
 
-    def _upload_fobj(self, fobj, to_info, **kwargs):
+    def upload_fobj(self, fobj, to_info, **kwargs):
         with self.hdfs_client.write(to_info.path) as fdest:
             shutil.copyfileobj(fobj, fdest)
 

dvc/machine/__init__.py

@@ -25,6 +25,18 @@ logger = logging.getLogger(__name__)
 
 RESERVED_NAMES = {"local", "localhost"}
 
+DEFAULT_STARTUP_SCRIPT = """#!/bin/bash
+sudo apt-get update
+sudo apt-get install --yes python3 python3-pip
+python3 -m pip install --upgrade pip
+
+pushd /etc/apt/sources.list.d
+sudo wget https://dvc.org/deb/dvc.list
+sudo apt-get update
+sudo apt-get install --yes dvc
+popd
+"""
+
 
 def validate_name(name: str):
     from dvc.exceptions import InvalidArgumentError
@@ -168,6 +180,12 @@ class MachineManager:
     def create(self, name: Optional[str]):
         """Create and start the specified machine instance."""
         config, backend = self.get_config_and_backend(name)
+        if "startup_script" in config:
+            with open(config["startup_script"]) as fobj:
+                startup_script = fobj.read()
+        else:
+            startup_script = DEFAULT_STARTUP_SCRIPT
+        config["startup_script"] = startup_script
         return backend.create(**config)
 
     def destroy(self, name: Optional[str]):

dvc/machine/backend/base.py

@@ -1,15 +1,38 @@
-from abc import abstractmethod
+from abc import ABC, abstractmethod
 from contextlib import contextmanager
 from typing import TYPE_CHECKING, Iterator, Optional
 
-from tpi import base
-
 if TYPE_CHECKING:
     from dvc.fs.ssh import SSHFileSystem
     from dvc.repo.experiments.executor.base import BaseExecutor
+    from dvc.types import StrPath
 
 
-class BaseMachineBackend(base.BaseMachineBackend):
+class BaseMachineBackend(ABC):
+    def __init__(self, tmp_dir: "StrPath", **kwargs):
+        raise NotImplementedError
+
+    @abstractmethod
+    def create(self, name: Optional[str] = None, **config):
+        """Create and start an instance of the specified machine."""
+
+    @abstractmethod
+    def destroy(self, name: Optional[str] = None, **config):
+        """Stop and destroy all instances of the specified machine."""
+
+    @abstractmethod
+    def instances(
+        self, name: Optional[str] = None, **config
+    ) -> Iterator[dict]:
+        """Iterate over status of all instances of the specified machine."""
+
+    def close(self):
+        pass
+
+    @abstractmethod
+    def run_shell(self, name: Optional[str] = None, **config):
+        """Spawn an interactive SSH shell for the specified machine."""
+
     @abstractmethod
     def get_executor(
         self, name: Optional[str] = None, **config

dvc/machine/backend/terraform.py

@@ -5,11 +5,17 @@ from tpi import TerraformProviderIterative, terraform
 
 from dvc.fs.ssh import SSHFileSystem
 
+from .base import BaseMachineBackend
+
 if TYPE_CHECKING:
     from dvc.repo.experiments.executor.base import BaseExecutor
+    from dvc.types import StrPath
 
 
-class TerraformBackend(terraform.TerraformBackend):
+class TerraformBackend(terraform.TerraformBackend, BaseMachineBackend):
+    def __init__(self, tmp_dir: "StrPath", **kwargs):
+        super().__init__(tmp_dir)
+
     def get_executor(
         self, name: Optional[str] = None, **config
     ) -> "BaseExecutor":

dvc/objects/db/reference.py

@@ -70,7 +70,7 @@ class ReferenceObjectDB(ObjectDB):
         ref_fobj = io.BytesIO(ref_file.to_bytes())
         ref_fobj.seek(0)
         try:
-            self.fs.upload_fobj(ref_fobj, to_info)
+            self.fs.upload(ref_fobj, to_info)
         except OSError as exc:
             if isinstance(exc, FileExistsError) or (
                 os.name == "nt"

dvc/objects/stage.py

@@ -34,9 +34,8 @@ def _upload_file(path_info, fs, odb, upload_odb):
     tmp_info = upload_odb.path_info / tmp_fname()
     with fs.open(path_info, mode="rb", chunk_size=fs.CHUNK_SIZE) as stream:
         stream = HashedStreamReader(stream)
-        upload_odb.fs.upload_fobj(
-            stream, tmp_info, desc=path_info.name, size=fs.getsize(path_info)
-        )
+        size = fs.getsize(path_info)
+        upload_odb.fs.upload(stream, tmp_info, desc=path_info.name, total=size)
 
     odb.add(tmp_info, upload_odb.fs, stream.hash_info)
     return path_info, odb.get(stream.hash_info)

dvc/objects/transfer.py

@@ -1,20 +1,19 @@
 import errno
-import itertools
 import logging
-from concurrent import futures
-from concurrent.futures import ThreadPoolExecutor
 from functools import partial, wraps
-from typing import TYPE_CHECKING, Callable, Iterable, Optional
+from typing import TYPE_CHECKING, Any, Callable, Iterable, Optional
 
 from funcy import split
 
 from dvc.progress import Tqdm
+from dvc.utils.threadpool import ThreadPoolExecutor
 
 if TYPE_CHECKING:
     from dvc.hash_info import HashInfo
 
     from .db.base import ObjectDB
     from .db.index import ObjectDBIndexBase
+    from .tree import Tree
 
 logger = logging.getLogger(__name__)
 
@@ -38,109 +37,48 @@ def _log_exceptions(func):
     return wrapper
 
 
-def _transfer(
-    src, dest, dir_ids, file_ids, missing_ids, jobs, verify, move, **kwargs
-):
-    func = _log_exceptions(dest.add)
-    total = len(dir_ids) + len(file_ids)
-    if total == 0:
-        return 0
-    with Tqdm(total=total, unit="file", desc="Transferring") as pbar:
-        func = pbar.wrap_fn(func)
-        with ThreadPoolExecutor(max_workers=jobs) as executor:
-            processor = partial(
-                _create_tasks,
-                executor,
-                jobs,
-                func,
-                src,
-                verify,
-                move,
-            )
-            processor.add_func = func
-            _do_transfer(
-                src,
-                dest,
-                dir_ids,
-                file_ids,
-                missing_ids,
-                processor,
-                verify=verify,
-                move=move,
-                **kwargs,
-            )
-    return total
+def find_tree_by_obj_id(
+    odbs: Iterable[Optional["ObjectDB"]], obj_id: "HashInfo"
+) -> Optional["Tree"]:
+    from .errors import ObjectFormatError
+    from .tree import Tree
 
-
-def _create_tasks(executor, jobs, func, src, verify, move, obj_ids):
-    fails = 0
-    hash_iter = iter(obj_ids)
-
-    def submit(hash_info):
-        obj = src.get(hash_info)
-        return executor.submit(
-            func,
-            obj.path_info,
-            obj.fs,
-            obj.hash_info,
-            verify=verify,
-            move=move,
-        )
-
-    def create_taskset(amount):
-        return {
-            submit(hash_info)
-            for hash_info in itertools.islice(hash_iter, amount)
-        }
-
-    tasks = create_taskset(jobs * 5)
-    while tasks:
-        done, tasks = futures.wait(tasks, return_when=futures.FIRST_COMPLETED)
-        fails += sum(task.result() for task in done)
-        tasks.update(create_taskset(len(done)))
-    return fails
+    for odb in odbs:
+        if odb is not None:
+            try:
+                return Tree.load(odb, obj_id)
+            except (FileNotFoundError, ObjectFormatError):
+                pass
+    return None
 
 
 def _do_transfer(
     src: "ObjectDB",
     dest: "ObjectDB",
-    dir_ids: Iterable["HashInfo"],
-    file_ids: Iterable["HashInfo"],
+    obj_ids: Iterable["HashInfo"],
     missing_ids: Iterable["HashInfo"],
     processor: Callable,
     src_index: Optional["ObjectDBIndexBase"] = None,
     dest_index: Optional["ObjectDBIndexBase"] = None,
     cache_odb: Optional["ObjectDB"] = None,
-    **kwargs,
+    **kwargs: Any,
 ):
     from dvc.exceptions import FileTransferError
-    from dvc.objects.errors import ObjectFormatError
 
+    dir_ids, file_ids = split(lambda hash_info: hash_info.isdir, obj_ids)
     total_fails = 0
     succeeded_dir_objs = []
     all_file_ids = set(file_ids)
 
     for dir_hash in dir_ids:
-        from .tree import Tree
-
-        bound_file_ids = set()
-        dir_obj: Optional["Tree"] = None
-        for odb in (cache_odb, src):
-            if odb is not None:
-                try:
-                    dir_obj = Tree.load(odb, dir_hash)
-                    break
-                except (FileNotFoundError, ObjectFormatError):
-                    pass
+        dir_obj = find_tree_by_obj_id([cache_odb, src], dir_hash)
         assert dir_obj
+
         entry_ids = {entry.hash_info for _, entry in dir_obj}
+        bound_file_ids = all_file_ids & entry_ids
+        all_file_ids -= entry_ids
 
-        for file_hash in all_file_ids.copy():
-            if file_hash in entry_ids:
-                bound_file_ids.add(file_hash)
-                all_file_ids.remove(file_hash)
-
-        dir_fails = processor(bound_file_ids)
+        dir_fails = sum(processor(bound_file_ids))
         if dir_fails:
             logger.debug(
                 "failed to upload full contents of '%s', "
@@ -164,19 +102,13 @@ def _do_transfer(
                 dir_obj.name,
             )
         else:
-            raw_obj = src.get(dir_obj.hash_info)
-            is_dir_failed = processor.add_func(  # type: ignore[attr-defined]
-                raw_obj.path_info,
-                raw_obj.fs,
-                raw_obj.hash_info,
-                **kwargs,
-            )
+            is_dir_failed = sum(processor([dir_obj.hash_info]))
             total_fails += is_dir_failed
             if not is_dir_failed:
                 succeeded_dir_objs.append(dir_obj)
 
     # insert the rest
-    total_fails += processor(all_file_ids)
+    total_fails += sum(processor(all_file_ids))
     if total_fails:
         if src_index:
             src_index.clear()
@@ -222,18 +154,19 @@ def transfer(
     if not status.new:
         return 0
 
-    dir_ids, file_ids = split(lambda hash_info: hash_info.isdir, status.new)
-    if jobs is None:
-        jobs = dest.fs.jobs
+    def func(hash_info: "HashInfo") -> None:
+        obj = src.get(hash_info)
+        return dest.add(
+            obj.path_info, obj.fs, obj.hash_info, verify=verify, move=move
+        )
 
-    return _transfer(
-        src,
-        dest,
-        set(dir_ids),
-        set(file_ids),
-        status.missing,
-        jobs,
-        verify,
-        move,
-        **kwargs,
-    )
+    total = len(status.new)
+    jobs = jobs or dest.fs.jobs
+    with Tqdm(total=total, unit="file", desc="Transferring") as pbar:
+        with ThreadPoolExecutor(max_workers=jobs) as executor:
+            wrapped_func = pbar.wrap_fn(_log_exceptions(func))
+            processor = partial(executor.imap_unordered, wrapped_func)
+            _do_transfer(
+                src, dest, status.new, status.missing, processor, **kwargs
+            )
+    return total

dvc/repo/diff.py

@@ -4,7 +4,6 @@ from collections import defaultdict
 from typing import Dict, List
 
 from dvc.exceptions import PathMissingError
-from dvc.objects.stage import get_file_hash
 from dvc.repo import locked
 from dvc.repo.experiments.utils import fix_exp_head
 
@@ -45,7 +44,7 @@ def diff(self, a_rev="HEAD", b_rev=None, targets=None):
                 repo_fs, targets
             )
 
-        results[rev] = _paths_checksums(self, repo_fs, targets_path_infos)
+        results[rev] = _paths_checksums(self, targets_path_infos)
 
     if targets is not None:
         # check for overlapping missing targets between a_rev and b_rev
@@ -98,7 +97,7 @@ def diff(self, a_rev="HEAD", b_rev=None, targets=None):
     return ret if any(ret.values()) else {}
 
 
-def _paths_checksums(repo, repo_fs, targets):
+def _paths_checksums(repo, targets):
     """
     A dictionary of checksums addressed by relpaths collected from
     the current fs outputs.
@@ -110,10 +109,10 @@ def _paths_checksums(repo, repo_fs, targets):
         file:      "data"
     """
 
-    return dict(_output_paths(repo, repo_fs, targets))
+    return dict(_output_paths(repo, targets))
 
 
-def _output_paths(repo, repo_fs, targets):
+def _output_paths(repo, targets):
     from dvc.fs.local import LocalFileSystem
     from dvc.objects.stage import stage as ostage
 
@@ -131,45 +130,47 @@ def _output_paths(repo, repo_fs, targets):
             else os.path.join(str(output), "")
         )
 
-    def _to_checksum(output):
-        if on_working_fs:
-            return ostage(
-                repo.odb.local,
-                output.path_info,
-                repo.odb.local.fs,
-                "md5",
-                dry_run=True,
-            )[1].hash_info.value
-        return output.hash_info.value
-
     for output in repo.index.outs:
         if _exists(output):
             yield_output = targets is None or any(
                 output.path_info.isin_or_eq(target) for target in targets
             )
 
+            if on_working_fs:
+                _, obj = ostage(
+                    repo.odb.local,
+                    output.path_info,
+                    repo.odb.local.fs,
+                    "md5",
+                    dry_run=True,
+                    dvcignore=output.dvcignore,
+                )
+                hash_info = obj.hash_info
+            else:
+                hash_info = output.hash_info
+                obj = output.get_obj()
+
             if yield_output:
-                yield _to_path(output), _to_checksum(output)
+                yield _to_path(output), hash_info.value
+
+            if not obj:
+                continue
 
             if output.is_dir_checksum and (
                 yield_output
                 or any(target.isin(output.path_info) for target in targets)
             ):
-                yield from _dir_output_paths(repo_fs, output, targets)
+                yield from _dir_output_paths(output.path_info, obj, targets)
 
 
-def _dir_output_paths(repo_fs, output, targets=None):
-    from dvc.config import NoRemoteError
-
-    try:
-        for fname in repo_fs.walk_files(output.path_info):
-            if targets is None or any(
-                fname.isin_or_eq(target) for target in targets
-            ):
-                # pylint: disable=no-member
-                yield str(fname), get_file_hash(fname, repo_fs, "md5").value
-    except NoRemoteError:
-        logger.warning("dir cache entry for '%s' is missing", output)
+def _dir_output_paths(path_info, obj, targets=None):
+    for key, entry_obj in obj:
+        fname = path_info.joinpath(*key)
+        if targets is None or any(
+            fname.isin_or_eq(target) for target in targets
+        ):
+            # pylint: disable=no-member
+            yield str(fname), entry_obj.hash_info.value
 
 
 def _filter_missing(repo_fs, paths):

dvc/scm/base.py

@@ -47,6 +47,15 @@ class InvalidRemoteSCMRepo(SCMError):
         super().__init__(msg)
 
 
+class GitAuthError(SCMError):
+    def __init__(self, url: str):
+        super().__init__(
+            f"HTTP Git authentication is not supported: '{url}'"
+            "\nSee https://dvc.org/doc//user-guide/"
+            "troubleshooting#git-auth"
+        )
+
+
 class Base:
     """Base class for source control management driver implementations."""
 

dvc/scm/git/backend/dulwich.py

@@ -20,7 +20,7 @@ from funcy import cached_property
 
 from dvc.path_info import PathInfo
 from dvc.progress import Tqdm
-from dvc.scm.base import InvalidRemoteSCMRepo, SCMError
+from dvc.scm.base import GitAuthError, InvalidRemoteSCMRepo, SCMError
 from dvc.utils import relpath
 
 from ..objects import GitObject
@@ -355,7 +355,7 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
                 yield os.fsdecode(key)
 
     def iter_remote_refs(self, url: str, base: Optional[str] = None):
-        from dulwich.client import get_transport_and_path
+        from dulwich.client import HTTPUnauthorized, get_transport_and_path
         from dulwich.errors import NotGitRepository
         from dulwich.porcelain import get_remote_repo
 
@@ -376,6 +376,8 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
                 yield from (os.fsdecode(ref) for ref in client.get_refs(path))
         except NotGitRepository as exc:
             raise InvalidRemoteSCMRepo(url) from exc
+        except HTTPUnauthorized:
+            raise GitAuthError(url)
 
     def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
         raise NotImplementedError
@@ -388,7 +390,7 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
         force: bool = False,
         on_diverged: Optional[Callable[[str, str], bool]] = None,
     ):
-        from dulwich.client import get_transport_and_path
+        from dulwich.client import HTTPUnauthorized, get_transport_and_path
         from dulwich.errors import NotGitRepository, SendPackError
         from dulwich.porcelain import (
             DivergedBranches,
@@ -447,6 +449,8 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
                 )
         except (NotGitRepository, SendPackError) as exc:
             raise SCMError("Git failed to push '{src}' to '{url}'") from exc
+        except HTTPUnauthorized:
+            raise GitAuthError(url)
 
     def _push_dest_refs(
         self, src: Optional[str], dest: str

dvc/state.py

@@ -53,7 +53,6 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
         self.tmp_dir = tmp_dir
         self.root_dir = root_dir
         self.dvcignore = dvcignore
-        self.fs = LocalFileSystem()
 
         if not tmp_dir:
             return
@@ -85,7 +84,7 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
         assert isinstance(hash_info, HashInfo)
         assert os.path.exists(path_info)
 
-        mtime, size = get_mtime_and_size(path_info, self.fs, self.dvcignore)
+        mtime, size = get_mtime_and_size(path_info, fs, self.dvcignore)
         inode = get_inode(path_info)
 
         logger.debug(
@@ -121,7 +120,7 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
         if not os.path.exists(path):
             return None
 
-        mtime, size = get_mtime_and_size(path, self.fs, self.dvcignore)
+        mtime, size = get_mtime_and_size(path, fs, self.dvcignore)
         inode = get_inode(path)
 
         value = self.md5s.get(inode)
@@ -143,10 +142,10 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
 
         assert isinstance(path_info, str) or path_info.scheme == "local"
 
-        if not self.fs.exists(path_info):
+        if not fs.exists(path_info):
             return
 
-        mtime, _ = get_mtime_and_size(path_info, self.fs, self.dvcignore)
+        mtime, _ = get_mtime_and_size(path_info, fs, self.dvcignore)
         inode = get_inode(path_info)
         relative_path = relpath(path_info, self.root_dir)
 
@@ -168,11 +167,11 @@ class State(StateBase):  # pylint: disable=too-many-instance-attributes
             for relative_path in ref:
                 path = os.path.join(self.root_dir, relative_path)
 
-                if path in used or not self.fs.exists(path):
+                if path in used or not fs.exists(path):
                     continue
 
                 inode = get_inode(path)
-                mtime, _ = get_mtime_and_size(path, self.fs, self.dvcignore)
+                mtime, _ = get_mtime_and_size(path, fs, self.dvcignore)
 
                 if ref[relative_path] == (inode, mtime):
                     logger.debug("Removing '%s' as unused link.", path)

dvc/utils/threadpool.py

@@ -0,0 +1,34 @@
+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
+
+    @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)))

dvc/version.py

@@ -6,7 +6,7 @@
 import os
 import subprocess
 
-_BASE_VERSION = "2.7.0"
+_BASE_VERSION = "2.7.3"
 
 
 def _generate_version(base_version):

requirements/azure.txt

@@ -0,0 +1,3 @@
+adlfs==2021.9.1
+azure-identity>=1.4.0
+knack

requirements/default.txt

@@ -0,0 +1,45 @@
+# See https://github.com/pyinstaller/pyinstaller/issues/1945
+ply>=3.9
+colorama>=0.3.9
+configobj>=5.0.6
+gitpython>3
+dulwich>=0.20.23
+pygit2>=1.5.0
+setuptools>=34.0.0
+nanotime>=0.5.2
+pyasn1>=0.4.1
+voluptuous>=0.11.7
+requests>=2.22.0
+grandalf>=0.6
+distro>=1.3.0
+appdirs>=1.4.3
+ruamel.yaml>=0.17.11
+toml>=0.10.1
+funcy>=1.14
+pathspec>=0.6.0,<0.9.0
+shortuuid>=0.5.0
+tqdm>=4.45.0,<5
+packaging>=19.0
+zc.lockfile>=1.2.1
+flufl.lock>=3.2,<4
+win-unicode-console>=0.5; sys_platform == 'win32'
+pywin32>=225; sys_platform == 'win32' and python_version < '3.10'
+networkx>=2.5
+psutil>=5.8.0
+pydot>=1.2.4
+speedcopy>=2.0.1; python_version < '3.8' and sys_platform == 'win32'
+dataclasses==0.7; python_version < '3.7'
+importlib-metadata>=1.4; python_version < '3.8'
+flatten_dict>=0.4.1,<1
+tabulate>=0.8.7
+pygtrie>=2.3.2
+dpath>=2.0.1,<3
+shtab>=1.3.4,<2
+rich>=10.0.0
+dictdiffer>=0.8.1
+python-benedict>=0.21.1
+pyparsing==2.4.7
+typing_extensions>=3.10.0.2
+fsspec[http]>=2021.8.1
+aiohttp-retry==2.4.5
+diskcache>=5.2.1

requirements/gdrive.txt

@@ -0,0 +1 @@
+pydrive2[fsspec]>=1.9.2

requirements/gs.txt

@@ -0,0 +1 @@
+gcsfs==2021.8.1

requirements/hdfs.txt

@@ -0,0 +1 @@
+pyarrow>=2.0.0; python_version < '3.10'

requirements/oss.txt

@@ -0,0 +1 @@
+ossfs==2021.8.0

requirements/s3.txt

@@ -0,0 +1,2 @@
+s3fs==2021.8.1
+aiobotocore[boto3]>1.0.1

requirements/ssh.txt

@@ -0,0 +1 @@
+sshfs>=2021.8.1

requirements/ssh_gssapi.txt

@@ -0,0 +1 @@
+sshfs[gssapi]>=2021.8.1

requirements/terraform.txt

@@ -0,0 +1 @@
+tpi[ssh]>=0.0.0

requirements/tests.txt

@@ -13,14 +13,13 @@ flaky==3.7.0
 mock==4.0.3
 rangehttpserver==1.2.0
 mock-ssh-server==0.9.0
-beautifulsoup4==4.9.3
 wget==3.2
 filelock==3.0.12
 wsgidav==3.1.1
 crc32c==2.2.post0
 xmltodict==0.12.0
 google-compute-engine==2.8.13
-google-cloud-storage==1.42.0
+google-cloud-storage==1.42.1
 urllib3==1.26.6
 dvclive>=0.2.1
 

requirements/webdav.txt

@@ -0,0 +1 @@
+webdav4>=0.9.1

requirements/webhdfs.txt

@@ -0,0 +1 @@
+hdfs==2.6.0

scripts/pyinstaller/hooks/hook-dvc.py

@@ -14,6 +14,8 @@ datas += copy_metadata("boto3")
 datas += copy_metadata("ossfs")
 datas += copy_metadata("sshfs")
 datas += copy_metadata("webdav4")
+datas += copy_metadata("aiohttp")
+datas += copy_metadata("aiohttp_retry")
 
 # https://github.com/pypa/setuptools/issues/1963
 hiddenimports = ["pkg_resources.py2_warn"]

setup.cfg

@@ -17,7 +17,7 @@ known_first_party=dvc,tests
 line_length=79
 
 [tool:pytest]
-timeout = 600
+timeout = 200
 timeout_method = thread
 log_level = debug
 addopts = -ra

setup.py

@@ -46,79 +46,25 @@ class build_py(_build_py):
         _build_py.run(self)
 
 
-install_requires = [
-    "ply>=3.9",  # See https://github.com/pyinstaller/pyinstaller/issues/1945
-    "colorama>=0.3.9",
-    "configobj>=5.0.6",
-    "gitpython>3",
-    "dulwich>=0.20.23",
-    "pygit2>=1.5.0",
-    "setuptools>=34.0.0",
-    "nanotime>=0.5.2",
-    "pyasn1>=0.4.1",
-    "voluptuous>=0.11.7",
-    "requests>=2.22.0",
-    "grandalf==0.6",
-    "distro>=1.3.0",
-    "appdirs>=1.4.3",
-    "ruamel.yaml>=0.17.11",
-    "toml>=0.10.1",
-    "funcy>=1.14",
-    "pathspec>=0.6.0,<0.9.0",
-    "shortuuid>=0.5.0",
-    "tqdm>=4.45.0,<5",
-    "packaging>=19.0",
-    "zc.lockfile>=1.2.1",
-    "flufl.lock>=3.2,<4",
-    "win-unicode-console>=0.5; sys_platform == 'win32'",
-    "pywin32>=225; sys_platform == 'win32' and python_version < '3.10'",
-    "networkx>=2.5",
-    "psutil>=5.8.0",
-    "pydot>=1.2.4",
-    "speedcopy>=2.0.1; python_version < '3.8' and sys_platform == 'win32'",
-    "dataclasses==0.7; python_version < '3.7'",
-    "importlib-metadata>=1.4; python_version < '3.8'",
-    "flatten_dict>=0.4.1,<1",
-    "tabulate>=0.8.7",
-    "pygtrie>=2.3.2",
-    "dpath>=2.0.1,<3",
-    "shtab>=1.3.4,<2",
-    "rich>=10.0.0",
-    "dictdiffer>=0.8.1",
-    "python-benedict>=0.21.1",
-    "pyparsing==2.4.7",
-    "typing_extensions>=3.7.4; python_version < '3.10'",
-    # until https://github.com/python/typing/issues/865 is fixed for python3.10
-    "typing_extensions==3.10.0.0; python_version >= '3.10'",
-    "fsspec>=2021.8.1",
-    "diskcache>=5.2.1",
-]
-
-
 # Extra dependencies for remote integrations
+requirements = {
+    path.stem: path.read_text().strip().splitlines()
+    for path in Path("requirements").glob("*.txt")
+}
 
-gs = ["gcsfs==2021.8.1"]
-gdrive = ["pydrive2[fsspec]>=1.9.2"]
-s3 = ["s3fs==2021.8.1", "aiobotocore[boto3]>1.0.1"]
-azure = ["adlfs==2021.8.2", "azure-identity>=1.4.0", "knack"]
-oss = ["ossfs==2021.8.0"]
-ssh = ["sshfs>=2021.8.1"]
-
-hdfs = ["pyarrow>=2.0.0; python_version < '3.10'"]
-webhdfs = ["hdfs==2.5.8"]
-webdav = ["webdav4>=0.9.0"]
 # gssapi should not be included in all_remotes, because it doesn't have wheels
 # for linux and mac, so it will fail to compile if user doesn't have all the
 # requirements, including kerberos itself. Once all the wheels are available,
 # we can start shipping it by default.
-ssh_gssapi = ["sshfs[gssapi]>=2021.8.1"]
-all_remotes = gs + s3 + azure + ssh + oss + gdrive + hdfs + webhdfs + webdav
 
-terraform = ["tpi[ssh]>=0.0.0"]
-
-tests_requirements = (
-    Path("test_requirements.txt").read_text().strip().splitlines()
-) + terraform
+install_requires = requirements.pop("default")
+requirements["all"] = [
+    requirements
+    for key, requirements in requirements.items()
+    if key not in ("tests", "ssh_gssapi", "terraform")
+]
+requirements["tests"] += requirements["terraform"]
+requirements["dev"] = requirements["all"] + requirements["tests"]
 
 setup(
     name="dvc",
@@ -132,21 +78,7 @@ setup(
     download_url="https://github.com/iterative/dvc",
     license="Apache License 2.0",
     install_requires=install_requires,
-    extras_require={
-        "all": all_remotes,
-        "gs": gs,
-        "gdrive": gdrive,
-        "s3": s3,
-        "azure": azure,
-        "oss": oss,
-        "ssh": ssh,
-        "ssh_gssapi": ssh_gssapi,
-        "hdfs": hdfs,
-        "webhdfs": webhdfs,
-        "webdav": webdav,
-        "terraform": terraform,
-        "tests": tests_requirements,
-    },
+    extras_require=requirements,
     keywords="data-science data-version-control machine-learning git"
     " developer-tools reproducibility collaboration ai",
     python_requires=">=3.6",

tests/func/experiments/conftest.py

@@ -88,3 +88,17 @@ def checkpoint_stage(tmp_dir, scm, dvc, mocker):
     scm.commit("init")
     stage.iterations = DEFAULT_ITERATIONS
     return stage
+
+
+@pytest.fixture
+def http_auth_patch(mocker):
+    from dulwich.client import HTTPUnauthorized
+
+    url = "https://0.0.0.0"
+    client = mocker.MagicMock()
+    client.get_refs.side_effect = HTTPUnauthorized("", url)
+    client.send_pack.side_effect = HTTPUnauthorized("", url)
+
+    patch = mocker.patch("dulwich.client.get_transport_and_path")
+    patch.return_value = (client, url)
+    return url

tests/func/experiments/test_remote.py

@@ -252,3 +252,37 @@ def test_push_pull_cache(
         path = os.path.join(dvc.odb.local.cache_dir, hash_[:2], hash_[2:])
         assert os.path.exists(path)
         assert open(path).read() == str(x)
+
+
+def test_auth_error_list(tmp_dir, scm, dvc, http_auth_patch):
+    from dvc.scm.base import GitAuthError
+
+    with pytest.raises(
+        GitAuthError,
+        match=f"HTTP Git authentication is not supported: '{http_auth_patch}'",
+    ):
+        dvc.experiments.ls(git_remote=http_auth_patch)
+
+
+def test_auth_error_pull(tmp_dir, scm, dvc, http_auth_patch):
+    from dvc.scm.base import GitAuthError
+
+    with pytest.raises(
+        GitAuthError,
+        match=f"HTTP Git authentication is not supported: '{http_auth_patch}'",
+    ):
+        dvc.experiments.pull(http_auth_patch, "foo")
+
+
+def test_auth_error_push(tmp_dir, scm, dvc, exp_stage, http_auth_patch):
+    from dvc.scm.base import GitAuthError
+
+    results = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
+    exp = first(results)
+    ref_info = first(exp_refs_by_rev(scm, exp))
+
+    with pytest.raises(
+        GitAuthError,
+        match=f"HTTP Git authentication is not supported: '{http_auth_patch}'",
+    ):
+        dvc.experiments.push(http_auth_patch, ref_info.name)

tests/func/experiments/test_show.py

@@ -9,6 +9,7 @@ from dvc.exceptions import InvalidArgumentError
 from dvc.main import main
 from dvc.repo.experiments.base import EXPS_STASH, ExpRefInfo
 from dvc.repo.experiments.executor.base import BaseExecutor, ExecutorInfo
+from dvc.repo.experiments.utils import exp_refs_by_rev
 from dvc.utils.fs import makedirs
 from dvc.utils.serialize import YAMLFileCorruptedError, dump_yaml
 from tests.func.test_repro_multistage import COPY_SCRIPT
@@ -269,6 +270,10 @@ def test_show_filter(
     included,
     excluded,
 ):
+    from contextlib import contextmanager
+
+    from dvc.ui import ui
+
     capsys.readouterr()
     div = "│" if os.name == "nt" else "┃"
 
@@ -312,13 +317,30 @@ def test_show_filter(
     if e_params is not None:
         command.append(f"--exclude-params={e_params}")
 
-    assert main(command) == 0
+    @contextmanager
+    def console_with(console, width):
+        console_options = console.options
+        original = console_options.max_width
+        con_width = console._width
+
+        try:
+            console_options.max_width = width
+            console._width = width
+            yield
+        finally:
+            console_options.max_width = original
+            console._width = con_width
+
+    with console_with(ui.rich_console, 255):
+        assert main(command) == 0
     cap = capsys.readouterr()
 
     for i in included:
-        assert f"{div} {i} {div}" in cap.out
+        assert f"{div} params.yaml:{i} {div}" in cap.out
+        assert f"{div} metrics.yaml:{i} {div}" in cap.out
     for e in excluded:
-        assert f"{div} {e} {div}" not in cap.out
+        assert f"{div} params.yaml:{e} {div}" not in cap.out
+        assert f"{div} metrics.yaml:{e} {div}" not in cap.out
 
 
 def test_show_multiple_commits(tmp_dir, scm, dvc, exp_stage):
@@ -412,7 +434,6 @@ def test_show_running_checkpoint(
 ):
     from dvc.repo.experiments.base import EXEC_BRANCH
     from dvc.repo.experiments.executor.local import TempDirExecutor
-    from dvc.repo.experiments.utils import exp_refs_by_rev
 
     baseline_rev = scm.get_rev()
     dvc.experiments.run(
@@ -471,3 +492,46 @@ def test_show_with_broken_repo(tmp_dir, scm, dvc, exp_stage, caplog):
 
     paths = ["workspace", "baseline", "error"]
     assert isinstance(get_in(result, paths), YAMLFileCorruptedError)
+
+
+def test_show_csv(tmp_dir, scm, dvc, exp_stage, capsys):
+    baseline_rev = scm.get_rev()
+
+    def _get_rev_isotimestamp(rev):
+        return datetime.fromtimestamp(
+            scm.gitpython.repo.rev_parse(rev).committed_date
+        ).isoformat()
+
+    result1 = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
+    rev1 = first(result1)
+    ref_info1 = first(exp_refs_by_rev(scm, rev1))
+    result2 = dvc.experiments.run(exp_stage.addressing, params=["foo=3"])
+    rev2 = first(result2)
+    ref_info2 = first(exp_refs_by_rev(scm, rev2))
+
+    capsys.readouterr()
+    assert main(["exp", "show", "--show-csv"]) == 0
+    cap = capsys.readouterr()
+    assert (
+        "Experiment,rev,typ,Created,parent,metrics.yaml:foo,params.yaml:foo"
+        in cap.out
+    )
+    assert ",workspace,baseline,,,3,3" in cap.out
+    assert (
+        "master,{},baseline,{},,1,1".format(
+            baseline_rev[:7], _get_rev_isotimestamp(baseline_rev)
+        )
+        in cap.out
+    )
+    assert (
+        "{},{},branch_base,{},,2,2".format(
+            ref_info1.name, rev1[:7], _get_rev_isotimestamp(rev1)
+        )
+        in cap.out
+    )
+    assert (
+        "{},{},branch_commit,{},,3,3".format(
+            ref_info2.name, rev2[:7], _get_rev_isotimestamp(rev2)
+        )
+        in cap.out
+    )

tests/func/test_fs.py

@@ -260,7 +260,6 @@ def test_fs_getsize(dvc, cloud):
         pytest.lazy_fixture("gs"),
         pytest.lazy_fixture("gdrive"),
         pytest.lazy_fixture("hdfs"),
-        pytest.lazy_fixture("http"),
         pytest.lazy_fixture("local_cloud"),
         pytest.lazy_fixture("oss"),
         pytest.lazy_fixture("s3"),
@@ -406,7 +405,7 @@ def test_fs_makedirs_on_upload_and_copy(dvc, cloud):
     fs = cls(**config)
 
     with io.BytesIO(b"foo") as stream:
-        fs.upload_fobj(stream, cloud / "dir" / "foo")
+        fs.upload(stream, cloud / "dir" / "foo")
 
     assert fs.isdir(cloud / "dir")
     assert fs.exists(cloud / "dir" / "foo")

tests/unit/command/test_experiments.py

@@ -1,3 +1,5 @@
+from datetime import datetime
+
 import pytest
 
 from dvc.cli import parse_args
@@ -12,6 +14,7 @@ from dvc.command.experiments import (
     CmdExperimentsRemove,
     CmdExperimentsRun,
     CmdExperimentsShow,
+    show_experiments,
 )
 from dvc.exceptions import InvalidArgumentError
 
@@ -277,3 +280,128 @@ def test_experiments_remove(dvc, scm, mocker, queue, clear_all, remote):
         clear_all=clear_all,
         remote="myremote" if remote else None,
     )
+
+
+all_experiments = {
+    "workspace": {
+        "baseline": {
+            "data": {
+                "timestamp": None,
+                "params": {
+                    "params.yaml": {
+                        "data": {
+                            "featurize": {"max_features": 3000, "ngrams": 1},
+                            "parent": 20170428,
+                            "train": {
+                                "n_est": 100,
+                                "min_split": 36,
+                            },
+                        }
+                    }
+                },
+                "queued": False,
+                "running": False,
+                "executor": None,
+                "metrics": {
+                    "scores.json": {
+                        "data": {
+                            "featurize": {"max_features": 3000, "ngrams": 1},
+                            "avg_prec": 0.5843640011189556,
+                            "roc_auc": 0.9544670443829399,
+                        }
+                    }
+                },
+            }
+        }
+    },
+    "b05eecc666734e899f79af228ff49a7ae5a18cc0": {
+        "baseline": {
+            "data": {
+                "timestamp": datetime(2021, 8, 2, 16, 48, 14),
+                "params": {
+                    "params.yaml": {
+                        "data": {
+                            "featurize": {"max_features": 3000, "ngrams": 1},
+                            "parent": 20170428,
+                            "train": {
+                                "n_est": 100,
+                                "min_split": 2,
+                            },
+                        }
+                    }
+                },
+                "queued": False,
+                "running": False,
+                "executor": None,
+                "metrics": {
+                    "scores.json": {
+                        "data": {
+                            "featurize": {"max_features": 3000, "ngrams": 1},
+                            "avg_prec": 0.5325162867864254,
+                            "roc_auc": 0.9106964878520005,
+                        }
+                    }
+                },
+                "name": "master",
+            }
+        },
+        "ae99936461d6c3092934160f8beafe66a294f98d": {
+            "data": {
+                "timestamp": datetime(2021, 8, 31, 14, 56, 55),
+                "params": {
+                    "params.yaml": {
+                        "data": {
+                            "featurize": {"max_features": 3000, "ngrams": 1},
+                            "parent": 20170428,
+                            "train": {
+                                "n_est": 100,
+                                "min_split": 36,
+                            },
+                        }
+                    }
+                },
+                "queued": True,
+                "running": True,
+                "executor": None,
+                "metrics": {
+                    "scores.json": {
+                        "data": {
+                            "featurize": {"max_features": 3000, "ngrams": 1},
+                            "avg_prec": 0.5843640011189556,
+                            "roc_auc": 0.9544670443829399,
+                        }
+                    }
+                },
+                "name": "exp-44136",
+            }
+        },
+    },
+}
+
+
+def test_show_experiments(capsys):
+    show_experiments(
+        all_experiments, precision=None, fill_value="", iso=True, show_csv=True
+    )
+    cap = capsys.readouterr()
+    assert (
+        "Experiment,rev,typ,Created,parent,State,scores.json:"
+        "featurize.max_features,scores.json:featurize.ngrams,"
+        "avg_prec,roc_auc,params.yaml:featurize.max_features,"
+        "params.yaml:featurize.ngrams,params.yaml:parent,"
+        "train.n_est,train.min_split" in cap.out
+    )
+    assert (
+        ",workspace,baseline,,,,3000,1,0.5843640011189556,0.9544670443829399,"
+        "3000,1,20170428,100,36" in cap.out
+    )
+    assert (
+        "master,b05eecc,baseline,2021-08-02T16:48:14,,,3000,1,"
+        "0.5325162867864254,0.9106964878520005,3000,1,20170428,100,2"
+        in cap.out
+    )
+    assert (
+        "exp-44136,ae99936,branch_base,2021-08-31T14:56:55,,Running,"
+        "3000,1,0.5843640011189556,0.9544670443829399,3000,1,20170428,100,36"
+        in cap.out
+    )

tests/unit/fs/test_ssh.py

@@ -34,10 +34,13 @@ def test_get_kwargs_from_urls():
 
 
 def test_init():
-    fs = SSHFileSystem(user="test", host="12.34.56.78", port="1234")
+    fs = SSHFileSystem(
+        user="test", host="12.34.56.78", port="1234", password="xxx"
+    )
     assert fs.fs_args["username"] == "test"
     assert fs.fs_args["host"] == "12.34.56.78"
     assert fs.fs_args["port"] == "1234"
+    assert fs.fs_args["password"] == fs.fs_args["passphrase"] == "xxx"
 
 
 mock_ssh_config = """

tests/unit/remote/test_http.py

@@ -1,17 +1,15 @@
-import io
+import ssl
 
 import pytest
-import requests
+from mock import patch
 
-from dvc.exceptions import HTTPError
 from dvc.fs.http import HTTPFileSystem
-from dvc.path_info import URLInfo
 
 
 def test_download_fails_on_error_code(dvc, http):
     fs = HTTPFileSystem(**http.config)
 
-    with pytest.raises(HTTPError):
+    with pytest.raises(FileNotFoundError):
         fs._download(http / "missing.txt", "missing.txt")
 
 
@@ -25,15 +23,13 @@ def test_public_auth_method(dvc):
 
     fs = HTTPFileSystem(**config)
 
-    assert fs._auth_method(config["url"]) is None
+    assert "auth" not in fs.fs_args["client_kwargs"]
+    assert "headers" not in fs.fs_args
 
 
 def test_basic_auth_method(dvc):
-    from requests.auth import HTTPBasicAuth
-
     user = "username"
     password = "password"
-    auth = HTTPBasicAuth(user, password)
     config = {
         "url": "http://example.com/",
         "path_info": "file.html",
@@ -44,28 +40,8 @@ def test_basic_auth_method(dvc):
 
     fs = HTTPFileSystem(**config)
 
-    assert fs._auth_method(config["url"]) == auth
-    assert isinstance(fs._auth_method(config["url"]), HTTPBasicAuth)
-
-
-def test_digest_auth_method(dvc):
-    from requests.auth import HTTPDigestAuth
-
-    user = "username"
-    password = "password"
-    auth = HTTPDigestAuth(user, password)
-    config = {
-        "url": "http://example.com/",
-        "path_info": "file.html",
-        "auth": "digest",
-        "user": user,
-        "password": password,
-    }
-
-    fs = HTTPFileSystem(**config)
-
-    assert fs._auth_method(config["url"]) == auth
-    assert isinstance(fs._auth_method(config["url"]), HTTPDigestAuth)
+    assert fs.fs_args["client_kwargs"]["auth"].login == user
+    assert fs.fs_args["client_kwargs"]["auth"].password == password
 
 
 def test_custom_auth_method(dvc):
@@ -81,17 +57,9 @@ def test_custom_auth_method(dvc):
 
     fs = HTTPFileSystem(**config)
 
-    assert fs._auth_method(config["url"]) is None
-    assert header in fs.headers
-    assert fs.headers[header] == password
-
-
-def test_ssl_verify_is_enabled_by_default(dvc):
-    config = {"url": "http://example.com/", "path_info": "file.html"}
-
-    fs = HTTPFileSystem(**config)
-
-    assert fs._session.verify is True
+    headers = fs.fs_args["headers"]
+    assert header in headers
+    assert headers[header] == password
 
 
 def test_ssl_verify_disable(dvc):
@@ -102,11 +70,11 @@ def test_ssl_verify_disable(dvc):
     }
 
     fs = HTTPFileSystem(**config)
-
-    assert fs._session.verify is False
+    assert not fs.fs_args["client_kwargs"]["connector"]._ssl
 
 
-def test_ssl_verify_custom_cert(dvc):
+@patch("ssl.SSLContext.load_verify_locations")
+def test_ssl_verify_custom_cert(dvc, mocker):
     config = {
         "url": "http://example.com/",
         "path_info": "file.html",
@@ -115,69 +83,19 @@ def test_ssl_verify_custom_cert(dvc):
 
     fs = HTTPFileSystem(**config)
 
-    assert fs._session.verify == "/path/to/custom/cabundle.pem"
+    assert isinstance(
+        fs.fs_args["client_kwargs"]["connector"]._ssl, ssl.SSLContext
+    )
 
 
 def test_http_method(dvc):
-    from requests.auth import HTTPBasicAuth
-
-    user = "username"
-    password = "password"
-    auth = HTTPBasicAuth(user, password)
     config = {
         "url": "http://example.com/",
         "path_info": "file.html",
-        "auth": "basic",
-        "user": user,
-        "password": password,
-        "method": "PUT",
     }
 
-    fs = HTTPFileSystem(**config)
+    fs = HTTPFileSystem(**config, method="PUT")
+    assert fs.upload_method == "PUT"
 
-    assert fs._auth_method(config["url"]) == auth
-    assert fs.method == "PUT"
-    assert isinstance(fs._auth_method(config["url"]), HTTPBasicAuth)
-
-
-def test_exists(mocker):
-    res = requests.Response()
-    # need to add `raw`, as `exists()` fallbacks to a streaming GET requests
-    # on HEAD request failure.
-    res.raw = io.StringIO("foo")
-
-    fs = HTTPFileSystem()
-    mocker.patch.object(fs, "request", return_value=res)
-
-    url = URLInfo("https://example.org/file.txt")
-
-    res.status_code = 200
-    assert fs.exists(url) is True
-
-    res.status_code = 404
-    assert fs.exists(url) is False
-
-    res.status_code = 403
-    with pytest.raises(HTTPError):
-        fs.exists(url)
-
-
-@pytest.mark.parametrize(
-    "headers, expected_size", [({"Content-Length": "3"}, 3), ({}, None)]
-)
-def test_content_length(mocker, headers, expected_size):
-    res = requests.Response()
-    res.headers.update(headers)
-    res.status_code = 200
-
-    fs = HTTPFileSystem()
-    mocker.patch.object(fs, "request", return_value=res)
-
-    url = URLInfo("https://example.org/file.txt")
-
-    assert fs.info(url) == {
-        "etag": None,
-        "size": expected_size,
-        "type": "file",
-    }
-    assert fs._content_length(res) == expected_size
+    fs = HTTPFileSystem(**config, method="POST")
+    assert fs.upload_method == "POST"