api: open/read: introduce remote_config

dvc/api/data.py

@@ -75,6 +75,7 @@ def open(  # noqa, pylint: disable=redefined-builtin
     mode: str = "r",
     encoding: Optional[str] = None,
     config: Optional[Dict[str, Any]] = None,
+    remote_config: Optional[Dict[str, Any]] = None,
 ):
     """
     Opens a file tracked in a DVC project.
@@ -118,6 +119,9 @@ def open(  # noqa, pylint: disable=redefined-builtin
             Mirrors the namesake parameter in builtin `open()`_.
         config(dict, optional): config to be passed to the DVC repository.
             Defaults to None.
+        remote_config(dict, optional): remote config to be passed to the DVC
+            repository.
+            Defaults to None.
 
     Returns:
         _OpenContextManager: A context manager that generatse a corresponding
@@ -214,22 +218,27 @@ def open(  # noqa, pylint: disable=redefined-builtin
         "mode": mode,
         "encoding": encoding,
         "config": config,
+        "remote_config": remote_config,
     }
     return _OpenContextManager(_open, args, kwargs)
 
 
-def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config=None):
-    if remote:
-        if config is not None:
-            raise ValueError(
-                "can't specify both `remote` and `config` at the same time"
-            )
-        config = {"core": {"remote": remote}}
-
+def _open(
+    path,
+    repo=None,
+    rev=None,
+    remote=None,
+    mode="r",
+    encoding=None,
+    config=None,
+    remote_config=None,
+):
     repo_kwargs: Dict[str, Any] = {
         "subrepos": True,
         "uninitialized": True,
+        "remote": remote,
         "config": config,
+        "remote_config": remote_config,
     }
 
     with Repo.open(repo, rev=rev, **repo_kwargs) as _repo:
@@ -265,7 +274,16 @@ def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None, confi
                 raise DvcIsADirectoryError(f"'{path}' is a directory") from exc
 
 
-def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config=None):
+def read(
+    path,
+    repo=None,
+    rev=None,
+    remote=None,
+    mode="r",
+    encoding=None,
+    config=None,
+    remote_config=None,
+):
     """
     Returns the contents of a tracked file (by DVC or Git). For Git repos, HEAD
     is used unless a rev argument is supplied. The default remote is tried
@@ -279,5 +297,6 @@ def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config
         mode=mode,
         encoding=encoding,
         config=config,
+        remote_config=remote_config,
     ) as fd:
         return fd.read()

dvc/config.py

@@ -93,6 +93,8 @@ class Config(dict):
         validate: bool = True,
         fs: Optional["FileSystem"] = None,
         config: Optional["DictStrAny"] = None,
+        remote: Optional[str] = None,
+        remote_config: Optional["DictStrAny"] = None,
     ):  # pylint: disable=super-init-not-called
         from dvc.fs import LocalFileSystem
 
@@ -103,7 +105,9 @@ class Config(dict):
         if dvc_dir:
             self.dvc_dir = self.fs.path.abspath(self.fs.path.realpath(dvc_dir))
 
-        self.load(validate=validate, config=config)
+        self.load(
+            validate=validate, config=config, remote=remote, remote_config=remote_config
+        )
 
     @classmethod
     def from_cwd(cls, fs: Optional["FileSystem"] = None, **kwargs):
@@ -154,7 +158,13 @@ class Config(dict):
         with open(config_file, "w+", encoding="utf-8"):
             return Config(dvc_dir)
 
-    def load(self, validate: bool = True, config: Optional["DictStrAny"] = None):
+    def load(
+        self,
+        validate: bool = True,
+        config: Optional["DictStrAny"] = None,
+        remote: Optional[str] = None,
+        remote_config: Optional["DictStrAny"] = None,
+    ):
         """Loads config from all the config files.
 
         Raises:
@@ -169,6 +179,17 @@ class Config(dict):
             conf = self.validate(conf)
 
         self.clear()
+
+        if remote:
+            conf["core"]["remote"] = remote
+
+        if remote_config:
+            remote = remote or conf["core"].get("remote")
+            if not remote:
+                raise ValueError("Missing remote name")
+
+            merge(conf, {"remote": {remote: remote_config}})
+
         self.update(conf)
 
     def _get_fs(self, level):

dvc/repo/__init__.py

@@ -133,7 +133,7 @@ class Repo:
         assert root_dir
         return root_dir, dvc_dir
 
-    def __init__(  # noqa: PLR0915
+    def __init__(  # noqa: PLR0915, PLR0913
         self,
         root_dir: Optional[str] = None,
         fs: Optional["FileSystem"] = None,
@@ -144,6 +144,8 @@ class Repo:
         url: Optional[str] = None,
         repo_factory: Optional[Callable] = None,
         scm: Optional[Union["Git", "NoSCM"]] = None,
+        remote: Optional[str] = None,
+        remote_config: Optional["DictStrAny"] = None,
     ):
         from dvc.cachemgr import CacheManager
         from dvc.data_cloud import DataCloud
@@ -163,6 +165,8 @@ class Repo:
         self._fs = fs or localfs
         self._scm = scm
         self._config = config
+        self._remote = remote
+        self._remote_config = remote_config
         self._data_index = None
 
         if rev and not fs:
@@ -237,7 +241,13 @@ class Repo:
     def config(self):
         from dvc.config import Config
 
-        return Config(self.dvc_dir, fs=self.fs, config=self._config)
+        return Config(
+            self.dvc_dir,
+            fs=self.fs,
+            config=self._config,
+            remote=self._remote,
+            remote_config=self._remote_config,
+        )
 
     @cached_property
     def local_dvc_dir(self):