Function _open

dvc/api/data.py:258–302

(
    path,
    repo=None,
    rev=None,
    remote=None,
    mode="r",
    encoding=None,
    config=None,
    remote_config=None,
)

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:
        with _wrap_exceptions(_repo, path):
            import os
            from typing import TYPE_CHECKING, Union

            from dvc.exceptions import IsADirectoryError as DvcIsADirectoryError
            from dvc.fs.data import DataFileSystem
            from dvc.fs.dvc import DVCFileSystem

            if TYPE_CHECKING:
                from dvc.fs import FileSystem

            fs: Union[FileSystem, DataFileSystem, DVCFileSystem]
            if os.path.isabs(path):
                fs = DataFileSystem(index=_repo.index.data["local"])
                fs_path = path
            else:
                fs = DVCFileSystem(repo=_repo, subrepos=True)
                fs_path = fs.from_os_path(path)

            try:
                with fs.open(fs_path, mode=mode, encoding=encoding) as fobj:
                    yield fobj
            except FileNotFoundError as exc:
                raise FileMissingError(path) from exc
            except IsADirectoryError as exc:
                raise DvcIsADirectoryError(f"'{path}' is a directory") from exc


def read(