pared: Test checks only pull. I am thinking about making the test child of ```TestDataCloudBase``` and using ```self.cloud.get(md5)``` to check more steps along the way. What do you think?
shcheklein: Please, don't forget to update docs or add a ticket to the dvc.org to update them
efiop: @pared Looks great! :fireworks: Indeed, need to test everything that uses this flag. Using TestDataCloudBase should work. 
diff --git a/dvc/cli.py b/dvc/cli.py
index 949a79bc5..7703e445a 100644
--- a/dvc/cli.py
+++ b/dvc/cli.py
@@ -355,6 +355,11 @@ def parse_args(argv=None):
                         default=False,
                         help='Do not prompt when removing '
                              'working directory files.')
+    pull_parser.add_argument(
+                        '-R',
+                        '--recursive',
+                        help='Pull cache for subdirectories of the specified '
+                             'directory.')
     pull_parser.set_defaults(func=CmdDataPull)
 
     # Push
@@ -387,6 +392,11 @@ def parse_args(argv=None):
                         default=False,
                         help='Push cache for all dependencies of the '
                              'specified target.')
+    push_parser.add_argument(
+                        '-R',
+                        '--recursive',
+                        help='Push cache from subdirectories of specified '
+                             'directory.')
     push_parser.set_defaults(func=CmdDataPush)
 
     # Fetch
@@ -419,6 +429,11 @@ def parse_args(argv=None):
                         default=False,
                         help='Fetch cache for all dependencies of the '
                              'specified target.')
+    fetch_parser.add_argument(
+                        '-R',
+                        '--recursive',
+                        help='Fetch cache for subdirectories of specified '
+                             'directory.')
     fetch_parser.set_defaults(func=CmdDataFetch)
 
     # Status
diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py
index 0fa44e162..31398d2d0 100644
--- a/dvc/command/data_sync.py
+++ b/dvc/command/data_sync.py
@@ -27,7 +27,8 @@ class CmdDataPull(CmdDataBase):
                               all_branches=self.args.all_branches,
                               all_tags=self.args.all_tags,
                               with_deps=self.args.with_deps,
-                              force=self.args.force)
+                              force=self.args.force,
+                              from_directory=self.args.recursive)
         except Exception:
             logger.error('failed to pull data from the cloud')
             return 1
@@ -43,7 +44,8 @@ class CmdDataPush(CmdDataBase):
                               show_checksums=self.args.show_checksums,
                               all_branches=self.args.all_branches,
                               all_tags=self.args.all_tags,
-                              with_deps=self.args.with_deps)
+                              with_deps=self.args.with_deps,
+                              from_directory=self.args.recursive)
         except Exception:
             logger.error('failed to push data to the cloud')
             return 1
@@ -59,7 +61,8 @@ class CmdDataFetch(CmdDataBase):
                                show_checksums=self.args.show_checksums,
                                all_branches=self.args.all_branches,
                                all_tags=self.args.all_tags,
-                               with_deps=self.args.with_deps)
+                               with_deps=self.args.with_deps,
+                               from_directory=self.args.recursive)
         except Exception:
             logger.error('failed to fetch data from the cloud')
             return 1
diff --git a/dvc/project.py b/dvc/project.py
index 04fb30a69..a3398680a 100644
--- a/dvc/project.py
+++ b/dvc/project.py
@@ -554,8 +554,12 @@ class Project(object):
                 used.append(out.path)
         self.state.remove_unused_links(used)
 
-    def checkout(self, target=None, with_deps=False, force=False):
-        all_stages = self.active_stages()
+    def checkout(self,
+                 target=None,
+                 with_deps=False,
+                 force=False,
+                 from_directory=None):
+        all_stages = self.active_stages(from_directory)
         stages = all_stages
 
         if target:
@@ -680,7 +684,8 @@ class Project(object):
                     all_tags=False,
                     remote=None,
                     force=False,
-                    jobs=None):
+                    jobs=None,
+                    from_directory=None):
         cache = {}
         cache['local'] = []
         cache['s3'] = []
@@ -695,9 +700,9 @@ class Project(object):
                 stages = self._collect(target,
                                        with_deps=with_deps)
             elif active:
-                stages = self.active_stages()
+                stages = self.active_stages(from_directory)
             else:
-                stages = self.stages()
+                stages = self.stages(from_directory)
 
             for stage in stages:
                 if active and not target and stage.locked:
@@ -820,7 +825,8 @@ class Project(object):
              all_branches=False,
              show_checksums=False,
              with_deps=False,
-             all_tags=False):
+             all_tags=False,
+             from_directory=None):
         with self.state:
             used = self._used_cache(target,
                                     all_branches=all_branches,
@@ -828,7 +834,8 @@ class Project(object):
                                     with_deps=with_deps,
                                     force=True,
                                     remote=remote,
-                                    jobs=jobs)['local']
+                                    jobs=jobs,
+                                    from_directory=from_directory)['local']
             self.cloud.push(used,
                             jobs,
                             remote=remote,
@@ -841,7 +848,8 @@ class Project(object):
               all_branches=False,
               show_checksums=False,
               with_deps=False,
-              all_tags=False):
+              all_tags=False,
+              from_directory=None):
         with self.state:
             used = self._used_cache(target,
                                     all_branches=all_branches,
@@ -849,7 +857,8 @@ class Project(object):
                                     with_deps=with_deps,
                                     force=True,
                                     remote=remote,
-                                    jobs=jobs)['local']
+                                    jobs=jobs,
+                                    from_directory=from_directory)['local']
             self.cloud.pull(used,
                             jobs,
                             remote=remote,
@@ -863,15 +872,20 @@ class Project(object):
              show_checksums=False,
              with_deps=False,
              all_tags=False,
-             force=False):
+             force=False,
+             from_directory=None):
         self.fetch(target,
                    jobs,
                    remote=remote,
                    all_branches=all_branches,
                    all_tags=all_tags,
                    show_checksums=show_checksums,
-                   with_deps=with_deps)
-        self.checkout(target=target, with_deps=with_deps, force=force)
+                   with_deps=with_deps,
+                   from_directory=from_directory)
+        self.checkout(target=target,
+                      with_deps=with_deps,
+                      force=force,
+                      from_directory=from_directory)
 
     def _local_status(self, target=None, with_deps=False):
         status = {}
@@ -1139,13 +1153,13 @@ class Project(object):
     def metrics_remove(self, path):
         self._metrics_modify(path, delete=True)
 
-    def graph(self):
+    def graph(self, from_directory=None):
         import networkx as nx
         from dvc.exceptions import OutputDuplicationError
 
         G = nx.DiGraph()
         G_active = nx.DiGraph()
-        stages = self.stages()
+        stages = self.stages(from_directory)
 
         outs = []
         outs_by_path = {}
@@ -1182,17 +1196,17 @@ class Project(object):
 
         return G, G_active
 
-    def pipelines(self):
+    def pipelines(self, from_directory=None):
         import networkx as nx
 
-        G, G_active = self.graph()
+        G, G_active = self.graph(from_directory)
 
         return [
             G.subgraph(c).copy()
             for c in nx.weakly_connected_components(G)
         ]
 
-    def stages(self):
+    def stages(self, from_directory=None):
         """
         Walks down the root directory looking for Dvcfiles,
         skipping the directories that are related with
@@ -1203,9 +1217,12 @@ class Project(object):
         """
         from dvc.stage import Stage
 
+        if not from_directory:
+            from_directory = self.root_dir
+
         stages = []
         outs = []
-        for root, dirs, files in os.walk(self.root_dir):
+        for root, dirs, files in os.walk(from_directory):
             for fname in files:
                 path = os.path.join(root, fname)
                 if not Stage.is_stage_file(path):
@@ -1228,11 +1245,11 @@ class Project(object):
 
         return stages
 
-    def active_stages(self):
+    def active_stages(self, from_directory=None):
         import networkx as nx
 
         stages = []
-        for G in self.pipelines():
+        for G in self.pipelines(from_directory):
             stages.extend(list(nx.get_node_attributes(G, 'stage').values()))
         return stages
 

pull/push/etc: support -R option
E.g. `dvc pull -R ./subfolder` would walk down the subfolder, find all the .dvc files and pull all of them.
diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py
index 2ba6ce17c..d13eb4040 100644
--- a/tests/test_data_cloud.py
+++ b/tests/test_data_cloud.py
@@ -15,6 +15,7 @@ from dvc.config import Config
 from dvc.data_cloud import (DataCloud, RemoteS3, RemoteGS, RemoteAzure,
                             RemoteLOCAL, RemoteSSH, RemoteHDFS, RemoteHTTP)
 from dvc.remote.base import STATUS_OK, STATUS_NEW, STATUS_DELETED
+from dvc.utils import file_md5
 
 from tests.basic_env import TestDvc
 
@@ -597,3 +598,89 @@ class TestWarnOnOutdatedStage(TestDvc):
         self.color_patch.start()
         self._test()
         self.color_patch.stop()
+
+
+class TestRecursiveSyncOperations(TestDataCloudBase):
+    def main(self, args):
+        ret = main(args)
+        self.assertEqual(ret, 0)
+
+    def _get_url(self):
+        self.dname = get_local_url()
+        return self.dname
+
+    def _should_test(self):
+        return True
+
+    def _get_cloud_class(self):
+        return RemoteLOCAL
+
+    def _prepare_repo(self):
+        self.main(['remote',
+                   'add',
+                   '-d',
+                   TEST_REMOTE,
+                   self.cloud._cloud.cache_dir])
+        self.create(self.DATA, self.DATA_CONTENTS)
+        self.create(self.DATA_SUB, self.DATA_SUB_CONTENTS)
+
+        self.dvc.add(self.DATA)
+        self.dvc.add(self.DATA_SUB_CONTENTS)
+
+    def _remove_local_data_files(self):
+        os.remove(self.DATA)
+        os.remove(self.DATA_SUB)
+
+    def _test_recursive_pull(self):
+        self._remove_local_data_files()
+        self._clear_local_cache()
+
+        self.assertFalse(os.path.exists(self.DATA))
+        self.assertFalse(os.path.exists(self.DATA_SUB))
+
+        self.main(['pull', '-R', self.DATA_DIR])
+
+        self.assertTrue(os.path.exists(self.DATA))
+        self.assertTrue(os.path.exists(self.DATA_SUB))
+
+    def _clear_local_cache(self):
+        shutil.rmtree(self.dvc.cache.local.cache_dir)
+
+    def _test_recursive_fetch(self, data_md5, data_sub_md5):
+        self._clear_local_cache()
+
+        local_cache_data_path = self.dvc.cache.local.get(data_md5)
+        local_cache_data_sub_path = self.dvc.cache.local.get(data_sub_md5)
+
+        self.assertFalse(os.path.exists(local_cache_data_path))
+        self.assertFalse(os.path.exists(local_cache_data_sub_path))
+
+        self.main(['fetch', '-R', self.DATA_DIR])
+
+        self.assertTrue(os.path.exists(local_cache_data_path))
+        self.assertTrue(os.path.exists(local_cache_data_sub_path))
+
+    def _test_recursive_push(self, data_md5, data_sub_md5):
+        cloud_data_path = self.cloud._cloud.get(data_md5)
+        cloud_data_sub_path = self.cloud._cloud.get(data_sub_md5)
+
+        self.assertFalse(os.path.exists(cloud_data_path))
+        self.assertFalse(os.path.exists(cloud_data_sub_path))
+
+        self.main(['push', '-R', self.DATA_DIR])
+
+        self.assertTrue(os.path.exists(cloud_data_path))
+        self.assertTrue(os.path.exists(cloud_data_sub_path))
+
+    def test(self):
+        self._setup_cloud()
+        self._prepare_repo()
+
+        data_md5 = file_md5(self.DATA)[0]
+        data_sub_md5 = file_md5(self.DATA_SUB)[0]
+
+        self._test_recursive_push(data_md5, data_sub_md5)
+
+        self._test_recursive_fetch(data_md5, data_sub_md5)
+
+        self._test_recursive_pull()

pip install -e .[all]
parse_log_pytest
requirements.txt
pytest>=3.3.1 nose>=1.3.7 mock>=2.0.0 coverage>=4.5.1 codecov>=2.0.15 xmltodict>=0.11.0 awscli>=1.14.18 google-compute-engine Pygments collective.checkdocs
apt-get update apt-get install -y gcc
3.6
requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
altgraph==0.17.4
appdirs==1.4.4
asciimatics==1.14.0
attrs==22.2.0
awscli==1.24.10
azure-common==1.1.28
azure-nspkg==3.0.2
azure-storage-blob==1.3.0
azure-storage-common==1.3.0
azure-storage-nspkg==3.1.0
bcrypt==4.0.1
boto==2.49.0
boto3==1.7.4
botocore==1.10.84
cachetools==4.2.4
certifi==2021.5.30
cffi==1.15.1
charset-normalizer==2.0.12
codecov==2.1.13
collective.checkdocs==0.2
colorama==0.4.4
configobj==5.0.8
configparser==5.2.0
coverage==6.2
cryptography==40.0.2
decorator==4.4.2
distro==1.9.0
docutils==0.16
-e git+https://github.com/iterative/dvc.git@a15f68a20decaf3a00b2d82a05eb88ce20a70b86#egg=dvc
future==1.0.0
gitdb==4.0.9
GitPython==3.1.18
google-api-core==1.32.0
google-auth==1.35.0
google-cloud-core==0.28.1
google-cloud-storage==1.13.0
google-compute-engine==2.8.13
google-crc32c==1.3.0
google-resumable-media==2.3.3
googleapis-common-protos==1.56.3
grandalf==0.6
idna==3.10
importlib-metadata==4.8.3
iniconfig==1.1.1
jmespath==0.10.0
jsonpath-rw==1.4.0
macholib==1.16.3
mock==5.2.0
nanotime==0.5.2
networkx==2.5.1
nose==1.3.7
packaging==21.3
paramiko==3.5.1
pefile==2024.8.26
Pillow==8.4.0
pluggy==1.0.0
ply==3.11
protobuf==3.19.6
py==1.11.0
pyasn1==0.5.1
pyasn1-modules==0.3.0
pycparser==2.21
pydot==1.4.2
pyfiglet==0.8.post1
Pygments==2.14.0
PyInstaller==3.3.1
PyNaCl==1.5.0
pyparsing==3.1.4
pytest==7.0.1
python-dateutil==2.9.0.post0
pytz==2025.2
PyYAML==5.4.1
requests==2.27.1
rsa==4.7.2
s3transfer==0.1.13
schema==0.7.7
six==1.17.0
smmap==5.0.0
tomli==1.2.3
typing_extensions==4.1.1
urllib3==1.26.20
wcwidth==0.2.13
xmltodict==0.14.2
zc.lockfile==2.0
zipp==3.6.0

name: dvc
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - altgraph==0.17.4
      - appdirs==1.4.4
      - asciimatics==1.14.0
      - attrs==22.2.0
      - awscli==1.24.10
      - azure-common==1.1.28
      - azure-nspkg==3.0.2
      - azure-storage-blob==1.3.0
      - azure-storage-common==1.3.0
      - azure-storage-nspkg==3.1.0
      - bcrypt==4.0.1
      - boto==2.49.0
      - boto3==1.7.4
      - botocore==1.10.84
      - cachetools==4.2.4
      - cffi==1.15.1
      - charset-normalizer==2.0.12
      - codecov==2.1.13
      - collective-checkdocs==0.2
      - colorama==0.4.4
      - configobj==5.0.8
      - configparser==5.2.0
      - coverage==6.2
      - cryptography==40.0.2
      - decorator==4.4.2
      - distro==1.9.0
      - docutils==0.16
      - future==1.0.0
      - gitdb==4.0.9
      - gitpython==3.1.18
      - google-api-core==1.32.0
      - google-auth==1.35.0
      - google-cloud-core==0.28.1
      - google-cloud-storage==1.13.0
      - google-compute-engine==2.8.13
      - google-crc32c==1.3.0
      - google-resumable-media==2.3.3
      - googleapis-common-protos==1.56.3
      - grandalf==0.6
      - idna==3.10
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - jmespath==0.10.0
      - jsonpath-rw==1.4.0
      - macholib==1.16.3
      - mock==5.2.0
      - nanotime==0.5.2
      - networkx==2.5.1
      - nose==1.3.7
      - packaging==21.3
      - paramiko==3.5.1
      - pefile==2024.8.26
      - pillow==8.4.0
      - pluggy==1.0.0
      - ply==3.11
      - protobuf==3.19.6
      - py==1.11.0
      - pyasn1==0.5.1
      - pyasn1-modules==0.3.0
      - pycparser==2.21
      - pydot==1.4.2
      - pyfiglet==0.8.post1
      - pygments==2.14.0
      - pyinstaller==3.3.1
      - pynacl==1.5.0
      - pyparsing==3.1.4
      - pytest==7.0.1
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - pyyaml==5.4.1
      - requests==2.27.1
      - rsa==4.7.2
      - s3transfer==0.1.13
      - schema==0.7.7
      - six==1.17.0
      - smmap==5.0.0
      - tomli==1.2.3
      - typing-extensions==4.1.1
      - urllib3==1.26.20
      - wcwidth==0.2.13
      - xmltodict==0.14.2
      - zc-lockfile==2.0
      - zipp==3.6.0
prefix: /opt/conda/envs/dvc

tests/test_data_cloud.py::TestRecursiveSyncOperations::test
tests/test_data_cloud.py::TestDataCloud::test tests/test_data_cloud.py::TestRemoteLOCAL::test tests/test_data_cloud.py::TestDataCloudCLIBase::test tests/test_data_cloud.py::TestCompatRemoteLOCALCLI::test tests/test_data_cloud.py::TestRemoteLOCALCLI::test tests/test_data_cloud.py::TestRemoteSSHCLI::test tests/test_data_cloud.py::TestRemoteHDFSCLI::test tests/test_data_cloud.py::TestCompatRemoteS3CLI::test tests/test_data_cloud.py::TestRemoteS3CLI::test tests/test_data_cloud.py::TestCompatRemoteGSCLI::test tests/test_data_cloud.py::TestRemoteGSCLI::test tests/test_data_cloud.py::TestRemoteAzureCLI::test tests/test_data_cloud.py::TestDataCloudErrorCLI::test_error tests/test_data_cloud.py::TestWarnOnOutdatedStage::test
dvc/command/data_sync.py dvc/project.py dvc/cli.py
dvc/command/data_sync.py dvc/project.py dvc/cli.py