Repository: getindata/data-pipelines-cli (language: python)

Issue:
Add `dp compile` and `dp deploy`
## Draft of `dp compile` and `dp deploy`

Depends on #2 PR.

---
Keep in mind:
- [ ] Documentation updates
- [X] [Changelog](CHANGELOG.md) updates


Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 12eecf4..6a9510d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,8 @@
 
 ## [Unreleased]
 
+- Add support for GCP and S3 syncing in `dp deploy`
+
 ## [0.1.2] - 2021-12-02
 
 -   Fix: do not use styled `click.secho` for Docker push response, as it may not be a `str`
diff --git a/README.md b/README.md
index 6fdce97..59d1068 100644
--- a/README.md
+++ b/README.md
@@ -10,6 +10,14 @@ First, run `dp init` to configure your first template. `dp init` will allow you
 
 `dp list` lists all added templates.
 
+When using `dp deploy`, it will sync with your bucket provider. Provider will be chosen automatically based on the
+remote URL. `dp deploy` also requires user to point to JSON or YAML file with provider-specific data like access tokens
+or project names. E.g., to connect with Google Cloud Storage, one should run:
+```bash
+echo '{"token": "<PATH_TO_YOUR_TOKEN>", "project_name": "<YOUR_PROJECT_NAME>"}' > gs_args.json
+dp deploy "gs://<YOUR_GS_PATH>" --blob-args gs_args.json
+```
+
 ## Contributing
 Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
 
diff --git a/data_pipelines_cli/cli_commands/deploy.py b/data_pipelines_cli/cli_commands/deploy.py
index 0b46680..95b065f 100644
--- a/data_pipelines_cli/cli_commands/deploy.py
+++ b/data_pipelines_cli/cli_commands/deploy.py
@@ -1,99 +1,129 @@
+import io
+import json
+import os
+import pathlib
 import subprocess
 import sys
-from typing import Optional
+from typing import Dict, Optional
 
 import click
+import yaml
 
-from ..cli_constants import BUILD_DIR, GS_BUCKET_ENV
-from ..cli_utils import echo_error, echo_info, get_argument_or_environment_variable
+from ..cli_constants import BUILD_DIR
+from ..cli_utils import echo_error, echo_info
 from ..data_structures import DockerArgs
-
-
-def _docker_push(docker_args: DockerArgs):
-    try:
-        import docker
-    except ModuleNotFoundError:
-        echo_error(
-            "'docker' not installed. Run 'pip install data-pipelines-cli[docker]'"
-        )
-        sys.exit(1)
-
-    echo_info("Pushing Docker image")
-    docker_client = docker.from_env()
-    for line in docker_client.images.push(
-        repository=docker_args.repository,
-        tag=docker_args.commit_sha,
-        stream=True,
+from ..filesystem_utils import LocalRemoteSync
+
+
+class DeployCommand:
+    docker_args: Optional[DockerArgs]
+    datahub_ingest: bool
+    blob_address_path: str
+    provider_kwargs_dict: Dict[str, str]
+
+    def __init__(
+        self,
+        repository: Optional[str],
+        blob_address: str,
+        provider_kwargs_dict: Dict[str, str],
+        docker_push: bool,
+        datahub_ingest: bool,
     ):
-        click.echo(line, nl=False)
-
-
-def _datahub_ingest():
-    try:
-        import datahub  # noqa: F401
-    except ModuleNotFoundError:
-        echo_error(
-            "'datahub' not installed. Run 'pip install data-pipelines-cli[datahub]'"
+        self.docker_args = DockerArgs(repository) if docker_push else None
+        self.datahub_ingest = datahub_ingest
+        self.blob_address_path = os.path.join(
+            blob_address, "dags", DeployCommand._get_project_name()
+        )
+        self.provider_kwargs_dict = provider_kwargs_dict
+
+    def deploy(self) -> None:
+        if self.docker_args:
+            self._docker_push()
+
+        if self.datahub_ingest:
+            self._datahub_ingest()
+
+        self._sync_bucket()
+
+    @staticmethod
+    def _get_project_name():
+        with open(pathlib.Path().joinpath("dbt_project.yml")) as f:
+            dbt_project_config = yaml.safe_load(f)
+            return dbt_project_config["name"]
+
+    def _docker_push(self):
+        try:
+            import docker
+        except ModuleNotFoundError:
+            echo_error(
+                "'docker' not installed. Run 'pip install data-pipelines-cli[docker]'"
+            )
+            sys.exit(1)
+
+        echo_info("Pushing Docker image")
+        docker_client = docker.from_env()
+        for line in docker_client.images.push(
+            repository=self.docker_args.repository,
+            tag=self.docker_args.commit_sha,
+            stream=True,
+        ):
+            click.echo(line, nl=False)
+
+    @staticmethod
+    def _datahub_ingest():
+        try:
+            import datahub  # noqa: F401
+        except ModuleNotFoundError:
+            echo_error(
+                "'datahub' not installed. Run 'pip install data-pipelines-cli[datahub]'"
+            )
+            sys.exit(1)
+
+        echo_info("Ingesting datahub metadata")
+        subprocess.run(
+            [
+                "datahub",
+                "ingest",
+                "-c",
+                str(BUILD_DIR.joinpath("dag", "config", "base", "datahub.yml")),
+            ]
         )
-        sys.exit(1)
-
-    echo_info("Ingesting datahub metadata")
-    subprocess.run(
-        [
-            "datahub",
-            "ingest",
-            "-c",
-            str(BUILD_DIR.joinpath("dag", "config", "base", "datahub.yml")),
-        ]
-    )
-    return True
-
-
-def _sync_bucket(gs_bucket: str) -> None:
-    build_dag_path = BUILD_DIR.joinpath("dag")
-
-    echo_info("Syncing GCP Bucket")
-    subprocess.run(
-        [
-            "gsutil",
-            "rsync",
-            "-d",
-            "-r",
-            str(build_dag_path),
-            f"{gs_bucket}/dags/pipeline-example",
-        ]
-    )
-
-
-def deploy(
-    repository: Optional[str],
-    gs_bucket: Optional[str],
-    docker_push: bool,
-    datahub_ingest: bool,
-) -> None:
-    gs_bucket = get_argument_or_environment_variable(
-        gs_bucket, "gs-bucket", GS_BUCKET_ENV
-    )
-
-    if docker_push:
-        docker_args = DockerArgs(repository)
-        _docker_push(docker_args)
-
-    if datahub_ingest:
-        _datahub_ingest()
 
-    _sync_bucket(gs_bucket)
+    def _sync_bucket(self) -> None:
+        echo_info("Syncing Bucket")
+        LocalRemoteSync(
+            BUILD_DIR.joinpath("dag"), self.blob_address_path, self.provider_kwargs_dict
+        ).sync(delete=True)
 
 
 @click.command(name="deploy")
+@click.argument("address")
+@click.option(
+    "--blob-args",
+    required=True,
+    type=click.File("r"),
+    help="Path to JSON or YAML file with arguments that should be passed to "
+    "your Bucket/blob provider",
+)
 @click.option("--repository", default=None)
-@click.option("--gs-bucket", default=None)
 @click.option("--docker-push", is_flag=True, default=False)
 @click.option("--datahub-ingest", is_flag=True, default=False)
 def deploy_command(
+    address: str,
+    blob_args: io.TextIOWrapper,
     repository: Optional[str],
-    gs_bucket: Optional[str],
     docker_push: bool,
     datahub_ingest: bool,
 ) -> None:
-    deploy(repository, gs_bucket, docker_push, datahub_ingest)
+    try:
+        provider_kwargs_dict = json.load(blob_args)
+    except json.JSONDecodeError:
+        provider_kwargs_dict = yaml.safe_load(blob_args)
+
+    DeployCommand(
+        repository,
+        address,
+        provider_kwargs_dict,
+        docker_push,
+        datahub_ingest,
+    ).deploy()
diff --git a/data_pipelines_cli/cli_constants.py b/data_pipelines_cli/cli_constants.py
index 4434318..6a0bf65 100644
--- a/data_pipelines_cli/cli_constants.py
+++ b/data_pipelines_cli/cli_constants.py
@@ -2,7 +2,6 @@
 
 CONFIGURATION_PATH: pathlib.Path = pathlib.Path.home().joinpath(".dp.yml")
 DATAHUB_URL_ENV: str = "DATAHUB_URL"
-GS_BUCKET_ENV: str = "GS_BUCKET"
 IMAGE_TAG_TO_REPLACE: str = "<IMAGE_TAG>"
 DOCKER_REPOSITORY_URL_TO_REPLACE: str = "<DOCKER_REPOSITORY_URL>"
 INGEST_ENDPOINT_TO_REPLACE: str = "<INGEST_ENDPOINT>"
diff --git a/data_pipelines_cli/filesystem_utils.py b/data_pipelines_cli/filesystem_utils.py
new file mode 100644
index 0000000..f667080
--- /dev/null
+++ b/data_pipelines_cli/filesystem_utils.py
@@ -0,0 +1,52 @@
+import os
+from typing import Dict, Optional, Set, Union
+
+import fsspec
+from fsspec import AbstractFileSystem
+
+
+# -------------------------------- Synchronize --------------------------------
+class LocalRemoteSync:
+    local_fs: AbstractFileSystem
+    local_path_str: str
+    remote_path_str: str
+    _local_directory_suffixes: Optional[Set[str]]
+
+    def __init__(
+        self,
+        local_path: Union[str, os.PathLike],
+        remote_path: str,
+        remote_kwargs: Dict[str, str],
+    ):
+        self.local_path_str = str(local_path).rstrip("/")
+        self.local_fs = fsspec.filesystem("file")
+        self.remote_fs, self.remote_path_str = fsspec.core.url_to_fs(
+            remote_path, **remote_kwargs
+        )
+        self._local_directory_suffixes = None
+
+    def sync(self, delete: bool = True):
+        self._push_sync()
+        if delete:
+            self._delete()
+
+    def _push_sync(self):
+        """Push every file to the remote"""
+
+        # TODO: Is it "lazy" (checking what to update) or not?
+        local_directory = self.local_fs.find(self.local_path_str)
+        self._local_directory_suffixes = set()
+        for local_file in local_directory:
+            local_file_suffix = local_file[len(self.local_path_str) :]
+            self._local_directory_suffixes.add(local_file_suffix)
+            self.remote_fs.put_file(
+                local_file, self.remote_path_str + local_file_suffix
+            )
+
+    def _delete(self):
+        """Remove every file from remote that's not local"""
+        remote_directory = self.remote_fs.find(self.remote_path_str)
+        for remote_file in remote_directory:
+            remote_file_suffix = remote_file[len(self.remote_path_str) :]
+            if remote_file_suffix not in self._local_directory_suffixes:
+                self.remote_fs.rm(remote_file)
diff --git a/setup.py b/setup.py
index 9a8229b..767a9d1 100644
--- a/setup.py
+++ b/setup.py
@@ -12,8 +12,14 @@
     "types-PyYAML>=6.0",
     "copier==5.1.0",
     "dbt>=0.21, <0.22",
+    "fsspec",
 ]
 
+EXTRA_FILESYSTEMS_REQUIRE = {
+    "gcp": ["gcsfs"],
+    "s3": ["s3fs"],
+}
+
 EXTRA_REQUIRE = {
     "docker": ["docker>=5.0"],
     "datahub": ["acryl-datahub>=0.8.17, <0.8.18"],
@@ -22,7 +28,16 @@
         "pytest-cov>=2.8.0, <3.0.0",
         "pre-commit==2.15.0",
         "tox==3.21.1",
+        "moto[s3]==2.2.16",
+        *(
+            [
+                require
+                for requires_list in EXTRA_FILESYSTEMS_REQUIRE.values()
+                for require in requires_list
+            ]
+        ),
     ],
+    **EXTRA_FILESYSTEMS_REQUIRE,
 }
 
 setup(


Test patch:
diff --git a/tests/test_filesystems.py b/tests/test_filesystems.py
new file mode 100644
index 0000000..01aee10
--- /dev/null
+++ b/tests/test_filesystems.py
@@ -0,0 +1,127 @@
+import pathlib
+import unittest
+
+import boto3
+import botocore
+import fsspec
+import moto
+from moto import mock_s3
+
+MY_BUCKET = "my_bucket"
+
+
+# According to
+# https://github.com/aio-libs/aiobotocore/issues/755#issuecomment-844273191
+# aiobotocore problems can be fixed by creating an AWSResponse with fixed
+# `raw_headers` field
+class MockedAWSResponse(botocore.awsrequest.AWSResponse):
+    raw_headers = {}  # type: ignore
+
+    async def read(self):  # type: ignore
+        return self.text
+
+
+class TestSynchronize(unittest.TestCase):
+    test_sync_2nd_directory_layout = [
+        "test2.txt",
+        str(pathlib.Path("a").joinpath("b", "c", "xyz")),
+    ]
+    test_sync_directory_layout = ["test1.txt", *test_sync_2nd_directory_layout]
+
+    def _test_synchronize(self, protocol: str, **remote_kwargs):
+        from data_pipelines_cli.filesystem_utils import LocalRemoteSync
+
+        local_path = pathlib.Path(__file__).parent.joinpath("test_sync_directory")
+        remote_path = f"{protocol}://{MY_BUCKET}/"
+        LocalRemoteSync(local_path, remote_path, remote_kwargs).sync(delete=False)
+
+        remote_fs, _ = fsspec.core.url_to_fs(remote_path)
+        for local_file in self.test_sync_directory_layout:
+            self.assertIn(
+                str(pathlib.Path(MY_BUCKET).joinpath(local_file)),
+                remote_fs.find(MY_BUCKET),
+            )
+
+    def _test_synchronize_with_delete(self, protocol: str, **remote_kwargs):
+        from data_pipelines_cli.filesystem_utils import LocalRemoteSync
+
+        local_path = pathlib.Path(__file__).parent.joinpath("test_sync_directory")
+        remote_path = f"{protocol}://{MY_BUCKET}/"
+        LocalRemoteSync(local_path, remote_path, remote_kwargs).sync(delete=True)
+
+        remote_fs, _ = fsspec.core.url_to_fs(remote_path)
+        for local_file in self.test_sync_directory_layout:
+            self.assertIn(
+                str(pathlib.Path(MY_BUCKET).joinpath(local_file)),
+                remote_fs.find(MY_BUCKET),
+            )
+
+        local_path_2 = pathlib.Path(__file__).parent.joinpath("test_sync_2nd_directory")
+        LocalRemoteSync(
+            local_path_2,
+            remote_path,
+            remote_kwargs,
+        ).sync(delete=True)
+        for local_file in self.test_sync_2nd_directory_layout:
+            self.assertIn(
+                str(pathlib.Path(MY_BUCKET).joinpath(local_file)),
+                remote_fs.find(MY_BUCKET),
+            )
+        self.assertNotIn(
+            str(pathlib.Path(MY_BUCKET).joinpath("test1.txt")),
+            remote_fs.find(MY_BUCKET),
+        )
+
+
+@mock_s3
+class TestS3Synchronize(TestSynchronize):
+    def setUp(self) -> None:
+        botocore.awsrequest.AWSResponse = MockedAWSResponse
+        moto.core.models.AWSResponse = MockedAWSResponse
+
+        client = boto3.client(
+            "s3",
+            region_name="eu-west-1",
+            aws_access_key_id="testing",
+            aws_secret_access_key="testing",
+        )
+        try:
+            s3 = boto3.resource(
+                "s3",
+                region_name="eu-west-1",
+                aws_access_key_id="testing",
+                aws_secret_access_key="testing",
+            )
+            s3.meta.client.head_bucket(Bucket=MY_BUCKET)
+        except botocore.exceptions.ClientError:
+            pass
+        else:
+            err = "{bucket} should not exist.".format(bucket=MY_BUCKET)
+            raise EnvironmentError(err)
+
+        client.create_bucket(
+            Bucket=MY_BUCKET,
+            CreateBucketConfiguration={"LocationConstraint": "eu-west-1"},
+        )
+
+    def tearDown(self):
+        s3 = boto3.resource(
+            "s3",
+            region_name="eu-west-1",
+            aws_access_key_id="testing",
+            aws_secret_access_key="testing",
+        )
+        bucket = s3.Bucket(MY_BUCKET)
+        for key in bucket.objects.all():
+            key.delete()
+        bucket.delete()
+
+    def test_synchronize(self):
+        self._test_synchronize("s3", key="testing", password="testing")
+
+    def test_synchronize_with_delete(self):
+        self._test_synchronize_with_delete("s3", key="testing", password="testing")
+
+
+# TODO: Although it would be nice to have unit tests for connecting with GCP,
+#  right now there's no elegant out-of-the-box mocking framework for GCS.
diff --git a/tests/test_sync_2nd_directory/a/b/c/xyz b/tests/test_sync_2nd_directory/a/b/c/xyz
new file mode 100644
index 0000000..e5d3534
--- /dev/null
+++ b/tests/test_sync_2nd_directory/a/b/c/xyz
@@ -0,0 +1,1 @@
+Lorem ipsum
\ No newline at end of file
diff --git a/tests/test_sync_2nd_directory/test2.txt b/tests/test_sync_2nd_directory/test2.txt
new file mode 100644
index 0000000..3b84bab
--- /dev/null
+++ b/tests/test_sync_2nd_directory/test2.txt
@@ -0,0 +1,9 @@
+Lorem ipsum dolor sit amet, consectetur adipiscing elit. Praesent commodo blandit justo. Maecenas tempor massa sit amet ante aliquet, sit amet condimentum eros vehicula. Proin consectetur ex at consectetur tristique. Morbi hendrerit sed ipsum eget elementum. Phasellus volutpat egestas congue. Sed ultrices rutrum mollis. Proin ac metus sed ligula congue blandit sit amet in erat. Ut pretium quam non malesuada blandit. Sed commodo lacus a neque fringilla, ac imperdiet lectus feugiat. Nam pulvinar, tortor vitae ullamcorper luctus, justo eros lobortis sem, ut ullamcorper turpis sapien nec quam.
+
+Fusce congue velit sit amet cursus ultrices. In hac habitasse platea dictumst. Nulla maximus efficitur quam at mattis. Phasellus nisl purus, consectetur in euismod et, finibus vel libero. Nulla facilisi. Quisque vehicula risus vitae purus cursus, in varius arcu pulvinar. Nam in neque congue, euismod felis id, volutpat lacus.
+
+Quisque vitae purus sit amet tortor dictum euismod nec non odio. Nam eu nulla augue. Donec vitae ipsum in libero mollis sagittis vel in turpis. Duis vel tempus nulla. Aenean in ullamcorper augue, eu volutpat nibh. Praesent interdum odio nisi, a faucibus diam finibus nec. Mauris ut lectus aliquet, dapibus leo vitae, pellentesque orci. Mauris neque arcu, condimentum nec lacus sed, mollis porta mi. Integer rhoncus congue risus, sed fermentum purus fringilla non. Duis pretium, odio vel vestibulum auctor, massa arcu iaculis enim, eu viverra nibh urna id massa. Mauris sagittis non justo ac tempor. Suspendisse non lacinia diam. Nullam in ante a risus consequat varius.
+
+Fusce vitae tempor ex. Duis tempus ligula a porttitor consequat. Praesent congue quam faucibus porta semper. Proin at pulvinar justo. Integer suscipit placerat nunc eget gravida. Nullam scelerisque, nisi nec gravida sollicitudin, orci metus fermentum ex, lacinia tempor risus lectus et metus. Sed et sapien at nisl feugiat malesuada. Praesent porttitor venenatis sem in ultrices.
+
+Aliquam at risus nec nulla ultricies laoreet. In pretium mattis libero, sed porta felis faucibus ac. Quisque eu tortor venenatis, pharetra eros volutpat, tempor nisi. Mauris sit amet massa ullamcorper, dapibus nisi et, ullamcorper mauris. Vivamus suscipit augue sapien, nec molestie ipsum venenatis quis. Donec ornare vehicula magna ut placerat. Aenean fermentum mi ac nunc consequat aliquet. Praesent mattis leo ut malesuada pulvinar. Aenean in libero leo. Etiam nec massa ut lectus efficitur ultrices id vel tellus. Vivamus lacinia interdum nisi eget volutpat. Ut elementum vehicula nunc, quis cursus eros ullamcorper et.
diff --git a/tests/test_sync_directory/a/b/c/xyz b/tests/test_sync_directory/a/b/c/xyz
new file mode 100644
index 0000000..e5d3534
--- /dev/null
+++ b/tests/test_sync_directory/a/b/c/xyz
@@ -0,0 +1,1 @@
+Lorem ipsum
\ No newline at end of file
diff --git a/tests/test_sync_directory/test1.txt b/tests/test_sync_directory/test1.txt
new file mode 100644
index 0000000..d96dc95
--- /dev/null
+++ b/tests/test_sync_directory/test1.txt
@@ -0,0 +1,1 @@
+abcdef
\ No newline at end of file
diff --git a/tests/test_sync_directory/test2.txt b/tests/test_sync_directory/test2.txt
new file mode 100644
index 0000000..3b84bab
--- /dev/null
+++ b/tests/test_sync_directory/test2.txt
@@ -0,0 +1,9 @@
+Lorem ipsum dolor sit amet, consectetur adipiscing elit. Praesent commodo blandit justo. Maecenas tempor massa sit amet ante aliquet, sit amet condimentum eros vehicula. Proin consectetur ex at consectetur tristique. Morbi hendrerit sed ipsum eget elementum. Phasellus volutpat egestas congue. Sed ultrices rutrum mollis. Proin ac metus sed ligula congue blandit sit amet in erat. Ut pretium quam non malesuada blandit. Sed commodo lacus a neque fringilla, ac imperdiet lectus feugiat. Nam pulvinar, tortor vitae ullamcorper luctus, justo eros lobortis sem, ut ullamcorper turpis sapien nec quam.
+
+Fusce congue velit sit amet cursus ultrices. In hac habitasse platea dictumst. Nulla maximus efficitur quam at mattis. Phasellus nisl purus, consectetur in euismod et, finibus vel libero. Nulla facilisi. Quisque vehicula risus vitae purus cursus, in varius arcu pulvinar. Nam in neque congue, euismod felis id, volutpat lacus.
+
+Quisque vitae purus sit amet tortor dictum euismod nec non odio. Nam eu nulla augue. Donec vitae ipsum in libero mollis sagittis vel in turpis. Duis vel tempus nulla. Aenean in ullamcorper augue, eu volutpat nibh. Praesent interdum odio nisi, a faucibus diam finibus nec. Mauris ut lectus aliquet, dapibus leo vitae, pellentesque orci. Mauris neque arcu, condimentum nec lacus sed, mollis porta mi. Integer rhoncus congue risus, sed fermentum purus fringilla non. Duis pretium, odio vel vestibulum auctor, massa arcu iaculis enim, eu viverra nibh urna id massa. Mauris sagittis non justo ac tempor. Suspendisse non lacinia diam. Nullam in ante a risus consequat varius.
+
+Fusce vitae tempor ex. Duis tempus ligula a porttitor consequat. Praesent congue quam faucibus porta semper. Proin at pulvinar justo. Integer suscipit placerat nunc eget gravida. Nullam scelerisque, nisi nec gravida sollicitudin, orci metus fermentum ex, lacinia tempor risus lectus et metus. Sed et sapien at nisl feugiat malesuada. Praesent porttitor venenatis sem in ultrices.
+
+Aliquam at risus nec nulla ultricies laoreet. In pretium mattis libero, sed porta felis faucibus ac. Quisque eu tortor venenatis, pharetra eros volutpat, tempor nisi. Mauris sit amet massa ullamcorper, dapibus nisi et, ullamcorper mauris. Vivamus suscipit augue sapien, nec molestie ipsum venenatis quis. Donec ornare vehicula magna ut placerat. Aenean fermentum mi ac nunc consequat aliquet. Praesent mattis leo ut malesuada pulvinar. Aenean in libero leo. Etiam nec massa ut lectus efficitur ultrices id vel tellus. Vivamus lacinia interdum nisi eget volutpat. Ut elementum vehicula nunc, quis cursus eros ullamcorper et.