Add CLI parameters for cdf data respace plan/execute

cognite_toolkit/_cdf_tk/apps/_respace_app.py

@@ -1,4 +1,5 @@
-from typing import Any
+from pathlib import Path
+from typing import Annotated, Any
 
 import typer
 from rich import print
@@ -20,13 +21,59 @@ def main(ctx: typer.Context) -> None:
             print("Use [bold yellow]cdf data respace --help[/] for more information.")
 
     @staticmethod
-    def respace_plan() -> None:
+    def respace_plan(
+        csv_file: Annotated[
+            Path,
+            typer.Argument(
+                help="Path to CSV file with nodes to respace. Expected columns: sourceSpace, externalId, targetSpace.",
+                exists=True,
+                file_okay=True,
+                dir_okay=False,
+                resolve_path=True,
+            ),
+        ],
+        output_file: Annotated[
+            Path,
+            typer.Option(
+                "--output-file",
+                "-o",
+                help="Path for the output plan file.",
+            ),
+        ] = Path("respace_plan.json"),
+    ) -> None:
         """Generate a respace plan by analyzing nodes and their dependencies."""
         cmd = RespaceCommand()
-        cmd.run(lambda: cmd.plan())
+        cmd.run(lambda: cmd.plan(csv_file=csv_file, output_file=output_file))
 
     @staticmethod
-    def respace_execute() -> None:
+    def respace_execute(
+        plan_file: Annotated[
+            Path,
+            typer.Argument(
+                help="Path to the respace plan JSON generated by 'plan' command.",
+                exists=True,
+                file_okay=True,
+                dir_okay=False,
+                resolve_path=True,
+            ),
+        ],
+        backup_dir: Annotated[
+            Path,
+            typer.Option(
+                "--backup-dir",
+                "-b",
+                help="Directory to store backup data before migration.",
+            ),
+        ],
+        dry_run: Annotated[
+            bool,
+            typer.Option(
+                "--dry-run",
+                "-d",
+                help="Preview changes without executing.",
+            ),
+        ] = False,
+    ) -> None:
         """Execute a respace plan, migrating nodes between spaces."""
         cmd = RespaceCommand()
-        cmd.run(lambda: cmd.execute())
+        cmd.run(lambda: cmd.execute(plan_file=plan_file, backup_dir=backup_dir, dry_run=dry_run))

cognite_toolkit/_cdf_tk/commands/_respace.py

@@ -1,10 +1,43 @@
+from pathlib import Path
+
+from cognite.client.utils._text import to_camel_case
+from pydantic import BaseModel, model_validator
 from rich import print
 
 from cognite_toolkit._cdf_tk.client import ToolkitClient
+from cognite_toolkit._cdf_tk.storageio._data_classes import ModelList
 
 from ._base import ToolkitCommand
 
 
+class RespaceMapping(BaseModel, alias_generator=to_camel_case, populate_by_name=True):
+    """A single node respace mapping from source to target.
+
+    The external ID is preserved — only the space changes.
+
+    CSV columns: sourceSpace, externalId, targetSpace
+    """
+
+    source_space: str
+    external_id: str
+    target_space: str
+
+    @model_validator(mode="after")
+    def _target_differs_from_source(self) -> "RespaceMapping":
+        if self.source_space == self.target_space:
+            raise ValueError(
+                f"targetSpace must differ from sourceSpace, "
+                f"but both are '{self.source_space}' for externalId '{self.external_id}'"
+            )
+        return self
+
+
+class RespaceMappingList(ModelList[RespaceMapping]):
+    @classmethod
+    def _get_base_model_cls(cls) -> type[RespaceMapping]:
+        return RespaceMapping
+
+
 class RespaceCommand(ToolkitCommand):
     def __init__(
         self,
@@ -15,10 +48,15 @@ def __init__(
     ):
         super().__init__(print_warning, skip_tracking, silent, client)
 
-    def plan(self) -> None:
+    def plan(self, csv_file: Path, output_file: Path) -> None:
         """Generate a respace plan from a CSV file."""
+        print(f"[bold]Planning respace from:[/] {csv_file}")
+        print(f"[bold]Output plan file:[/] {output_file}")
         print("[bold yellow]:construction: Work in Progress, you'll be able to plan soon! :construction:[/]")
 
-    def execute(self) -> None:
+    def execute(self, plan_file: Path, backup_dir: Path, dry_run: bool = False) -> None:
         """Execute a respace plan."""
+        verb = "Would execute" if dry_run else "Executing"
+        print(f"[bold]{verb} plan:[/] {plan_file}")
+        print(f"[bold]Backup directory:[/] {backup_dir}")
         print("[bold yellow]:construction: Work in Progress, you'll be able to execute soon! :construction:[/]")