ls/import/get: introduce --config

Adds support for `--config` to all three commands. For `ls` and `get`, it is
a purely runtime option that will merge the config you provide with target
repo's configs (similar to how .dvc/config.local works). E.g.

```
$ cat myconfig
[remote "myremote"]
    access_key_id = 12345
    secret_access_key = 98765

$ dvc get https://github.com/efiop/mydataregistry recent-grads.csv --config myconfig
```

In case of `dvc import`, the semantics are the same, but the value is also
recorded in the resulting dvcfile. E.g.

```
$ dvc import https://github.com/efiop/mydataregistry recent-grads.csv --config myconfig
...
$ cat recent-grads.csv.dvc
md5: 85c86eeb3afa6d1f7d11beedc644d28e
frozen: true
deps:
- path: recent-grads.csv
  repo:
    url: https://github.com/efiop/mydataregistry
    config: myconfig
    rev_lock: af22e06bbfe34d62a140437abf32f48b535944a7
outs:
- md5: f447e23a170a9f25a7799f069a1ba934
  size: 26872
  hash: md5
  path: recent-grads.csv
```

This is the most general and powerful mechanism that we can give, that works
for everything from reconfiguring your default remote to completely changing
your remote layout and beyond (will be handy in the future).

Fixes #2466

dvc/commands/get.py

@@ -37,6 +37,7 @@ def _get_file_from_repo(self):
                 rev=self.args.rev,
                 jobs=self.args.jobs,
                 force=self.args.force,
+                config=self.args.config,
             )
             return 0
         except CloneError:
@@ -102,4 +103,12 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Override local file or folder if exists.",
     )
+    get_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository.",
+        ),
+    )
     get_parser.set_defaults(func=CmdGet)

dvc/commands/imp.py

@@ -22,6 +22,7 @@ def run(self):
                 no_exec=self.args.no_exec,
                 no_download=self.args.no_download,
                 jobs=self.args.jobs,
+                config=self.args.config,
             )
         except CloneError:
             logger.exception("failed to import '%s'", self.args.path)
@@ -94,4 +95,12 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
+    import_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository.",
+        ),
+    )
     import_parser.set_defaults(func=CmdImport)

dvc/commands/ls/__init__.py

@@ -34,6 +34,7 @@ def run(self):
                 rev=self.args.rev,
                 recursive=self.args.recursive,
                 dvc_only=self.args.dvc_only,
+                config=self.args.config,
             )
             if self.args.json:
                 ui.write_json(entries)
@@ -80,6 +81,14 @@ def add_parser(subparsers, parent_parser):
         help="Git revision (e.g. SHA, branch, tag)",
         metavar="<commit>",
     )
+    list_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository.",
+        ),
+    )
     list_parser.add_argument(
         "path",
         nargs="?",

dvc/config.py

@@ -176,23 +176,32 @@ def _get_fs(self, level):
         # the repo.
         return self.fs if level == "repo" else self.wfs
 
-    def _load_config(self, level):
+    @staticmethod
+    def load_file(path, fs=None) -> dict:
         from configobj import ConfigObj, ConfigObjError
 
+        from dvc.fs import localfs
+
+        fs = fs or localfs
+
+        with fs.open(path) as fobj:
+            try:
+                conf_obj = ConfigObj(fobj)
+            except UnicodeDecodeError as exc:
+                raise ConfigError(str(exc)) from exc
+            except ConfigObjError as exc:
+                raise ConfigError(str(exc)) from exc
+
+        return _parse_named(_lower_keys(conf_obj.dict()))
+
+    def _load_config(self, level):
         filename = self.files[level]
         fs = self._get_fs(level)
 
-        if fs.exists(filename):
-            with fs.open(filename) as fobj:
-                try:
-                    conf_obj = ConfigObj(fobj)
-                except UnicodeDecodeError as exc:
-                    raise ConfigError(str(exc)) from exc
-                except ConfigObjError as exc:
-                    raise ConfigError(str(exc)) from exc
-        else:
-            conf_obj = ConfigObj()
-        return _parse_named(_lower_keys(conf_obj.dict()))
+        try:
+            return self.load_file(filename, fs=fs)
+        except FileNotFoundError:
+            return {}
 
     def _save_config(self, level, conf_dict):
         from configobj import ConfigObj

dvc/dependency/repo.py

@@ -16,12 +16,14 @@ class RepoDependency(Dependency):
     PARAM_URL = "url"
     PARAM_REV = "rev"
     PARAM_REV_LOCK = "rev_lock"
+    PARAM_CONFIG = "config"
 
     REPO_SCHEMA = {
         PARAM_REPO: {
             Required(PARAM_URL): str,
             PARAM_REV: str,
             PARAM_REV_LOCK: str,
+            PARAM_CONFIG: str,
         }
     }
 
@@ -77,9 +79,16 @@ def changed_checksum(self) -> bool:
     def _make_fs(
         self, rev: Optional[str] = None, locked: bool = True
     ) -> "DVCFileSystem":
+        from dvc.config import Config
         from dvc.fs import DVCFileSystem
 
-        config = {"cache": self.repo.config["cache"]}
+        conf = self.def_repo.get("config")
+        if conf:
+            config = Config.load_file(conf)
+        else:
+            config = {}
+
+        config["cache"] = self.repo.config["cache"]
         config["cache"]["dir"] = self.repo.cache.local_cache_dir
 
         return DVCFileSystem(

dvc/repo/get.py

@@ -19,7 +19,8 @@ def __init__(self):
         )
 
 
-def get(url, path, out=None, rev=None, jobs=None, force=False):
+def get(url, path, out=None, rev=None, jobs=None, force=False, config=None):
+    from dvc.config import Config
     from dvc.dvcfile import is_valid_filename
     from dvc.fs.callbacks import Callback
     from dvc.repo import Repo
@@ -29,11 +30,15 @@ def get(url, path, out=None, rev=None, jobs=None, force=False):
     if is_valid_filename(out):
         raise GetDVCFileError
 
+    if config and not isinstance(config, dict):
+        config = Config.load_file(config)
+
     with Repo.open(
         url=url,
         rev=rev,
         subrepos=True,
         uninitialized=True,
+        config=config,
     ) as repo:
         from dvc.fs.data import DataFileSystem
 

dvc/repo/imp.py

@@ -1,5 +1,5 @@
-def imp(self, url, path, out=None, rev=None, **kwargs):
-    erepo = {"url": url}
+def imp(self, url, path, out=None, rev=None, config=None, **kwargs):
+    erepo = {"url": url, "config": config}
     if rev is not None:
         erepo["rev"] = rev
 

dvc/repo/ls.py

@@ -13,6 +13,7 @@ def ls(
     rev: Optional[str] = None,
     recursive: Optional[bool] = None,
     dvc_only: bool = False,
+    config: Optional[str] = None,
 ):
     """Methods for getting files and outputs for the repo.
 
@@ -22,6 +23,7 @@ def ls(
         rev (str, optional): SHA commit, branch or tag name
         recursive (bool, optional): recursively walk the repo
         dvc_only (bool, optional): show only DVC-artifacts
+        config (bool, optional): path to config file
 
     Returns:
         list of `entry`
@@ -35,9 +37,18 @@ def ls(
             "isexec": bool,
         }
     """
+    from dvc.config import Config
+
     from . import Repo
 
-    with Repo.open(url, rev=rev, subrepos=True, uninitialized=True) as repo:
+    if config and not isinstance(config, dict):
+        config_dict = Config.load_file(config)
+    else:
+        config_dict = None
+
+    with Repo.open(
+        url, rev=rev, subrepos=True, uninitialized=True, config=config_dict
+    ) as repo:
         path = path or ""
 
         ret = _ls(repo, path, recursive, dvc_only)