api: add config support for open/read

Fixes #9610

dvc/api/data.py

@@ -73,6 +73,7 @@ def open(  # noqa, pylint: disable=redefined-builtin
     remote: Optional[str] = None,
     mode: str = "r",
     encoding: Optional[str] = None,
+    config: Optional[Dict[str, Any]] = None,
 ):
     """
     Opens a file tracked in a DVC project.
@@ -114,6 +115,8 @@ def open(  # noqa, pylint: disable=redefined-builtin
             Defaults to None.
             This should only be used in text mode.
             Mirrors the namesake parameter in builtin `open()`_.
+        config(dict, optional): config to be passed to the DVC repository.
+            Defaults to None.
 
     Returns:
         _OpenContextManager: A context manager that generatse a corresponding
@@ -209,14 +212,24 @@ def open(  # noqa, pylint: disable=redefined-builtin
         "rev": rev,
         "mode": mode,
         "encoding": encoding,
+        "config": config,
     }
     return _OpenContextManager(_open, args, kwargs)
 
 
-def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None):
-    repo_kwargs: Dict[str, Any] = {"subrepos": True, "uninitialized": True}
+def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config=None):
     if remote:
-        repo_kwargs["config"] = {"core": {"remote": remote}}
+        if config is not None:
+            raise ValueError(
+                "can't specify both `remote` and `config` at the same time"
+            )
+        config = {"core": {"remote": remote}}
+
+    repo_kwargs: Dict[str, Any] = {
+        "subrepos": True,
+        "uninitialized": True,
+        "config": config,
+    }
 
     with Repo.open(repo, rev=rev, **repo_kwargs) as _repo:
         with _wrap_exceptions(_repo, path):
@@ -251,13 +264,19 @@ def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None):
                 raise DvcIsADirectoryError(f"'{path}' is a directory") from exc
 
 
-def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None):
+def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None, 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
     unless a remote argument is supplied.
     """
     with open(
-        path, repo=repo, rev=rev, remote=remote, mode=mode, encoding=encoding
+        path,
+        repo=repo,
+        rev=rev,
+        remote=remote,
+        mode=mode,
+        encoding=encoding,
+        config=config,
     ) as fd:
         return fd.read()

tests/func/api/test_data.py

@@ -218,3 +218,36 @@ def test_open_from_remote(tmp_dir, erepo_dir, cloud, local_cloud):
         remote="other",
     ) as fd:
         assert fd.read() == "foo content"
+
+    with api.open(
+        os.path.join("dir", "foo"),
+        repo=f"file://{erepo_dir.as_posix()}",
+        config={"core": {"remote": "other"}},
+    ) as fd:
+        assert fd.read() == "foo content"
+
+
+def test_read_from_remote(tmp_dir, erepo_dir, cloud, local_cloud):
+    erepo_dir.add_remote(config=cloud.config, name="other")
+    erepo_dir.add_remote(config=local_cloud.config, default=True)
+    erepo_dir.dvc_gen({"dir": {"foo": "foo content"}}, commit="create file")
+    erepo_dir.dvc.push(remote="other")
+    remove(erepo_dir.dvc.cache.local.path)
+
+    assert (
+        api.read(
+            os.path.join("dir", "foo"),
+            repo=f"file://{erepo_dir.as_posix()}",
+            remote="other",
+        )
+        == "foo content"
+    )
+
+    assert (
+        api.read(
+            os.path.join("dir", "foo"),
+            repo=f"file://{erepo_dir.as_posix()}",
+            config={"core": {"remote": "other"}},
+        )
+        == "foo content"
+    )