[ACIX-1225] Create dda env dev fs import and export commands

src/dda/cli/env/dev/fs/__init__.py

@@ -0,0 +1,13 @@
+# SPDX-FileCopyrightText: 2024-present Datadog, Inc. <[email protected]>
+#
+# SPDX-License-Identifier: MIT
+from __future__ import annotations
+
+from dda.cli.base import dynamic_group
+
+
+@dynamic_group(
+    short_help="Interact with the environment's filesystem",
+)
+def cmd() -> None:
+    pass

src/dda/cli/env/dev/fs/export/__init__.py

@@ -0,0 +1,73 @@
+# SPDX-FileCopyrightText: 2024-present Datadog, Inc. <[email protected]>
+#
+# SPDX-License-Identifier: MIT
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+from dda.cli.base import dynamic_command, pass_app
+from dda.cli.env.dev.utils import option_env_type
+from dda.utils.fs import Path
+
+if TYPE_CHECKING:
+    from dda.cli.application import Application
+
+
+@dynamic_command(
+    short_help="""Export files and directories from a developer environment""",
+)
+@option_env_type()
+@click.option("--id", "instance", default="default", help="Unique identifier for the environment")
+@click.argument("sources", nargs=-1, required=True)
+@click.argument("destination", required=True, type=click.Path(resolve_path=True, path_type=Path))
+@click.option("--recursive", "-r", is_flag=True, help="Export files and directories recursively.")
+@click.option(
+    "--force",
+    "-f",
+    is_flag=True,
+    help="Overwrite existing files. Without this option, an error will be raised if the destination file already exists.",
+)
+@click.option(
+    "--mkpath", is_flag=True, help="Create the destination directories and their parents if they do not exist."
+)
+@pass_app
+def cmd(
+    app: Application,
+    *,
+    env_type: str,
+    instance: str,
+    sources: tuple[str, ...],  # Passed as string since they are inside the env filesystem
+    destination: Path,
+    recursive: bool,
+    force: bool,
+    mkpath: bool,
+) -> None:
+    """
+    Export files and directories from a developer environment, using an interface similar to `cp`.
+    The last path specified is the destination directory on the host filesystem.
+    """
+    from dda.env.dev import get_dev_env
+    from dda.env.models import EnvironmentState
+
+    env = get_dev_env(env_type)(
+        app=app,
+        name=env_type,
+        instance=instance,
+    )
+    status = env.status()
+
+    # TODO: This might end up depending on the environment type.
+    # For `linux-container` though, `docker cp` also works on stopped containers.
+    possible_states = {EnvironmentState.STARTED, EnvironmentState.STOPPED}
+    if status.state not in possible_states:
+        app.abort(
+            f"Developer environment `{env_type}` is in state `{status.state}`, must be one of: "
+            f"{', '.join(sorted(possible_states))}"
+        )
+
+    try:
+        env.export_files(sources, destination, recursive, force, mkpath)
+    except Exception as error:  # noqa: BLE001
+        app.abort(f"Failed to export files: {error}")

src/dda/cli/env/dev/fs/import/__init__.py

@@ -0,0 +1,69 @@
+# SPDX-FileCopyrightText: 2024-present Datadog, Inc. <[email protected]>
+#
+# SPDX-License-Identifier: MIT
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+from dda.cli.base import dynamic_command, pass_app
+from dda.cli.env.dev.utils import option_env_type
+from dda.utils.fs import Path
+
+if TYPE_CHECKING:
+    from dda.cli.application import Application
+
+
+@dynamic_command(short_help="""Import files and directories into a developer environment""")
+@option_env_type()
+@click.option("--id", "instance", default="default", help="Unique identifier for the environment")
+@click.argument("sources", nargs=-1, required=True, type=click.Path(exists=True, resolve_path=True, path_type=Path))
+@click.argument("destination", required=True)
+@click.option("--recursive", "-r", is_flag=True, help="Import files and directories recursively.")
+@click.option(
+    "--force",
+    "-f",
+    is_flag=True,
+    help="Overwrite existing files. Without this option, an error will be raised if the destination file already exists.",
+)
+@click.option(
+    "--mkpath", is_flag=True, help="Create the destination directories and their parents if they do not exist."
+)
+@pass_app
+def cmd(
+    app: Application,
+    *,
+    env_type: str,
+    instance: str,
+    sources: tuple[Path, ...],
+    destination: str,  # Passed as string since it is inside the env filesystem
+    recursive: bool,
+    force: bool,
+    mkpath: bool,
+) -> None:
+    """
+    Import files and directories into a developer environment, using an interface similar to `cp`.
+    The last path specified is the destination directory inside the environment.
+    """
+    from dda.env.dev import get_dev_env
+    from dda.env.models import EnvironmentState
+
+    env = get_dev_env(env_type)(
+        app=app,
+        name=env_type,
+        instance=instance,
+    )
+    status = env.status()
+
+    possible_states = {EnvironmentState.STARTED}
+    if status.state not in possible_states:
+        app.abort(
+            f"Developer environment `{env_type}` is in state `{status.state}`, must be one of: "
+            f"{', '.join(sorted(possible_states))}"
+        )
+
+    try:
+        env.import_files(sources, destination, recursive, force, mkpath)
+    except Exception as error:  # noqa: BLE001
+        app.abort(f"Failed to import files: {error}")

src/dda/cli/env/dev/fs/localimport/__init__.py

@@ -0,0 +1,35 @@
+# SPDX-FileCopyrightText: 2024-present Datadog, Inc. <[email protected]>
+#
+# SPDX-License-Identifier: MIT
+from __future__ import annotations
+
+import click
+
+from dda.cli.base import dynamic_command
+from dda.utils.fs import Path
+
+
+@dynamic_command(short_help="""Internal command used to call import_from_dir in dev envs.""", hidden=True)
+@click.argument(
+    "source", required=True, type=click.Path(exists=True, resolve_path=True, file_okay=False, path_type=Path)
+)
+@click.argument("destination", required=True, type=click.Path(resolve_path=True, path_type=Path))
+# Use arguments instead of options to enforce the idea that these are required
+@click.argument("recursive", required=True, type=bool)
+@click.argument("force", required=True, type=bool)
+@click.argument("mkpath", required=True, type=bool)
+def cmd(
+    *,
+    source: Path,
+    destination: Path,
+    recursive: bool,
+    force: bool,
+    mkpath: bool,
+) -> None:
+    """
+    Internal command used to call import_from_dir in dev envs.
+    This allows us to use the same semantics for importing files and directories into a dev env as for exporting them on the host filesystem.
+    """
+    from dda.env.dev.fs import import_from_dir
+
+    import_from_dir(source, destination, recursive=recursive, force=force, mkpath=mkpath)

src/dda/env/dev/fs.py

@@ -0,0 +1,81 @@
+# SPDX-FileCopyrightText: 2024-present Datadog, Inc. <[email protected]>
+#
+# SPDX-License-Identifier: MIT
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from dda.utils.fs import Path
+
+
+def determine_final_copy_target(source_name: str, source_is_dir: bool, destination_spec: Path) -> Path:  # noqa: FBT001
+    """
+    Determines the final target for a copy operation, given a destination specification and some details about the source.
+    For example:
+    - f("file.txt", False, "/tmp/some-dir") -> "/tmp/some-dir/file.txt" (move into directory)
+    - f("file.txt", False, "/tmp/new-file.txt") -> "/tmp/new-file.txt" (rename file)
+    - f("some-dir", True, "/tmp/some-dir") -> "/tmp/some-dir/some-dir" (move directory into directory)
+
+    Parameters:
+    - source_name: The name of the source file or directory. The source is usually inside the env filesystem, not the host.
+    - source_is_dir: Whether the source is a directory.
+    - destination_spec: The destination specification, which can be a directory or a file. The destination is usually on the host filesystem.
+
+    Returns:
+    - The final target path.
+    """
+
+    if destination_spec.is_dir():
+        # The destination exists and is a directory or a symlink to one
+        # Always move the source inside it
+        # TODO: Add a check if destination_spec / source.name is an already-existing file or directory
+        # Currently shutil.move will fail with an ugly error message when we eventually call it
+        return destination_spec / source_name
+
+    if destination_spec.is_file():
+        # The destination exists and is a file
+        if source_is_dir:
+            # Never overwrite a file with a directory
+            msg = f"Refusing to overwrite existing file with directory: {destination_spec}"
+            raise ValueError(msg)
+        # Source and destination are both files - rename
+        return destination_spec
+
+    # The destination does not exist, assume we want it exactly there
+    return destination_spec
+
+
+def handle_overwrite(dest: Path, *, force: bool) -> None:
+    if not dest.exists():
+        return
+
+    if dest.is_dir():
+        msg = f"Refusing to overwrite directory {dest}."
+        raise ValueError(msg)
+
+    if not force:
+        msg = f"Refusing to overwrite existing file: {dest} (force flag is not set)."
+        raise ValueError(msg)
+
+    dest.unlink()
+
+
+def import_from_dir(source_dir: Path, destination_spec: Path, *, recursive: bool, force: bool, mkpath: bool) -> None:
+    """
+    Import files and directories from a given directory into a destination directory on the "host" filesystem.
+    "Host" in this context refers to the environment `dda` is being executed in: if that is inside of a dev env, then we mean the dev env's file system.
+    """
+    from shutil import move
+
+    if mkpath:
+        destination_spec.ensure_dir()
+
+    for element in source_dir.iterdir():
+        if not recursive and element.is_dir():
+            msg = "Refusing to copy directories as recursive flag is not set"
+            raise ValueError(msg)
+
+        final_target = determine_final_copy_target(element.name, element.is_dir(), destination_spec)
+        handle_overwrite(final_target, force=force)
+        move(element, final_target)

src/dda/env/dev/interface.py

@@ -163,6 +163,48 @@ def launch_shell(self, *, repo: str | None = None) -> NoReturn:
                 [configured repository][dda.env.dev.interface.DeveloperEnvironmentConfig.repos].
         """
 
+    @abstractmethod
+    def export_files(
+        self,
+        sources: tuple[str, ...],  # Passed as string since they are inside the env filesystem
+        destination: Path,
+        recursive: bool,  # noqa: FBT001
+        force: bool,  # noqa: FBT001
+        mkpath: bool,  # noqa: FBT001
+    ) -> None:
+        """
+        This method exports files from the developer environment to the host filesystem.
+
+        Parameters:
+            sources: The paths to files/directories in the developer environment to export.
+            destination: The destination directory on the host filesystem.
+            recursive: Whether to export files and directories recursively. If False, all sources must be files.
+            force: Whether to overwrite existing files. Without this option, an error will be raised if the destination file/directory already exists.
+            mkpath: Whether to create the destination directories and their parents if they do not exist.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def import_files(
+        self,
+        sources: tuple[Path, ...],
+        destination: str,  # Passed as string since it is inside the env filesystem
+        recursive: bool,  # noqa: FBT001
+        force: bool,  # noqa: FBT001
+        mkpath: bool,  # noqa: FBT001
+    ) -> None:
+        """
+        This method imports files from the host filesystem into the developer environment.
+
+        Parameters:
+            sources: The paths to files/directories on the host filesystem to import.
+            destination: The destination directory in the developer environment.
+            recursive: Whether to import files and directories recursively. If False, all sources must be files.
+            force: Whether to overwrite existing files. Without this option, an error will be raised if the destination file/directory already exists.
+            mkpath: Whether to create the destination directories and their parents if they do not exist.
+        """
+        raise NotImplementedError
+
     def launch_gui(self) -> NoReturn:
         """
         This method starts an interactive GUI inside the developer environment using e.g. RDP or VNC.