import-url: pass fs_config down from imp_url to get_cloud_fs

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):

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)
 

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)
 

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)

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 [])
 
 

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"}