dvc: bump to 2.7.2

* Show csv format of experiments

fix #5989 and #5446 

1. add --show-csv to dvc exp show
2. add a new functional test for csv format
3. add a new unit test for show_experiements
4. solve name duplication in table common headers (Created,Parent, etc... ), metrics files and param files.

Co-authored-by: Peter Rowlands (변기호) <[email protected]>

.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 }}

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,

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/fsspec_wrapper.py

@@ -233,6 +233,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/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/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/version.py

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

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

@@ -90,7 +90,8 @@ install_requires = [
     "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",
+    "fsspec[http]>=2021.8.1",
+    "aiohttp-retry==2.4.5",
     "diskcache>=5.2.1",
 ]
 
@@ -106,7 +107,7 @@ ssh = ["sshfs>=2021.8.1"]
 
 hdfs = ["pyarrow>=2.0.0; python_version < '3.10'"]
 webhdfs = ["hdfs==2.5.8"]
-webdav = ["webdav4>=0.9.0"]
+webdav = ["webdav4>=0.9.1"]
 # 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,

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"),

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/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"