diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index b21bdf25e..8962dc302 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -18,7 +18,7 @@ SCHEMA: Mapping[str, Any] = { } -def _get(stage, p, info): +def _get(stage, p, info, fs_config=None): if info and info.get(RepoDependency.PARAM_REPO): repo = info.pop(RepoDependency.PARAM_REPO) return RepoDependency(repo, stage, p, info) @@ -27,7 +27,7 @@ def _get(stage, p, info): params = info.pop(ParamsDependency.PARAM_PARAMS) return ParamsDependency(stage, p, params) - return Dependency(stage, p, info) + return Dependency(stage, p, info, fs_config=fs_config) def loadd_from(stage, d_list): @@ -38,10 +38,18 @@ def loadd_from(stage, d_list): return ret -def loads_from(stage, s_list, erepo=None): +def loads_from(stage, s_list, erepo=None, fs_config=None): assert isinstance(s_list, list) info = {RepoDependency.PARAM_REPO: erepo} if erepo else {} - return [_get(stage, s, info.copy()) for s in s_list] + return [ + _get( + stage, + s, + info.copy(), + fs_config=fs_config, + ) + for s in s_list + ] def _merge_params(s_list): diff --git a/dvc/output.py b/dvc/output.py index 42c326130..9bc305dcf 100644 --- a/dvc/output.py +++ b/dvc/output.py @@ -290,6 +290,7 @@ class Output: meta=None, remote=None, repo=None, + fs_config=None, ): self.annot = Annotation( desc=desc, type=type, labels=labels or [], meta=meta or {} @@ -299,10 +300,12 @@ class Output: # NOTE: when version_aware is not passed into get_cloud_fs, it will be # set based on whether or not path is versioned fs_kwargs = {"version_aware": True} if meta.version_id else {} + + if fs_config is not None: + fs_kwargs.update(**fs_config) + fs_cls, fs_config, fs_path = get_cloud_fs( - self.repo, - url=path, - **fs_kwargs, + self.repo, url=path, **fs_kwargs ) self.fs = fs_cls(**fs_config) diff --git a/dvc/repo/imp_url.py b/dvc/repo/imp_url.py index d3c1e2390..66a8b8ac2 100644 --- a/dvc/repo/imp_url.py +++ b/dvc/repo/imp_url.py @@ -26,6 +26,7 @@ def imp_url( labels=None, meta=None, jobs=None, + fs_config=None, ): from dvc.dvcfile import Dvcfile from dvc.stage import Stage, create_stage, restore_fields @@ -61,6 +62,7 @@ def imp_url( deps=[url], outs=[out], erepo=erepo, + fs_config=fs_config, ) restore_fields(stage) diff --git a/dvc/stage/__init__.py b/dvc/stage/__init__.py index a0fa670b8..4dd5a923c 100644 --- a/dvc/stage/__init__.py +++ b/dvc/stage/__init__.py @@ -88,7 +88,7 @@ def create_stage(cls, repo, path, external=False, **kwargs): if not external: check_no_externals(stage) fill_stage_dependencies( - stage, **project(kwargs, ["deps", "erepo", "params"]) + stage, **project(kwargs, ["deps", "erepo", "params", "fs_config"]) ) check_circular_dependency(stage) check_duplicated_arguments(stage) diff --git a/dvc/stage/utils.py b/dvc/stage/utils.py index 2e2b66d24..6db4694a8 100644 --- a/dvc/stage/utils.py +++ b/dvc/stage/utils.py @@ -108,12 +108,16 @@ def _load_live_output( return outs -def fill_stage_dependencies(stage, deps=None, erepo=None, params=None): +def fill_stage_dependencies( + stage, deps=None, erepo=None, params=None, fs_config=None +): from dvc.dependency import loads_from, loads_params assert not stage.deps stage.deps = [] - stage.deps += loads_from(stage, deps or [], erepo=erepo) + stage.deps += loads_from( + stage, deps or [], erepo=erepo, fs_config=fs_config + ) stage.deps += loads_params(stage, params or []) diff --git a/tests/func/test_import_url.py b/tests/func/test_import_url.py index 6e34de8b3..84af3abeb 100644 --- a/tests/func/test_import_url.py +++ b/tests/func/test_import_url.py @@ -7,7 +7,7 @@ import pytest from dvc.annotations import Annotation from dvc.cli import main -from dvc.dependency.base import DependencyDoesNotExistError +from dvc.dependency.base import Dependency, DependencyDoesNotExistError from dvc.exceptions import InvalidArgumentError from dvc.stage import Stage from dvc.testing.test_workspace import TestImport as _TestImport @@ -330,3 +330,21 @@ def test_imp_url_with_annotations(M, tmp_dir, dvc, local_workspace): ) assert stage.outs[0].annot == Annotation(**annot) assert (tmp_dir / "foo.dvc").parse() == M.dict(outs=[M.dict(**annot)]) + + +def test_import_url_fs_config(tmp_dir, dvc, workspace, mocker): + import dvc.fs as dvc_fs + + workspace.gen("foo", "foo") + + url = "remote://workspace/foo" + get_fs_config = mocker.spy(dvc_fs, "get_fs_config") + dep_init = mocker.spy(Dependency, "__init__") + dvc.imp_url(url, fs_config={"jobs": 42}) + + dep_init_kwargs = dep_init.call_args[1] + assert dep_init_kwargs.get("fs_config") == {"jobs": 42} + + assert get_fs_config.call_args_list[0][1] == {"url": "foo"} + assert get_fs_config.call_args_list[1][1] == {"url": url, "jobs": 42} + assert get_fs_config.call_args_list[2][1] == {"name": "workspace"}