Dev

docker-compose-dev.yaml

@@ -1,7 +1,6 @@
 # Basic Airflow cluster configuration for CeleryExecutor with Redis and PostgreSQL.
 # WARNING: This configuration is for local development. Do not use it in a production deployment.
 ---
-version: '3.8'
 x-airflow-common:
   &airflow-common
   build: 

frontend/src/context/workspaces/workspaces.tsx

@@ -233,6 +233,11 @@ export const WorkspacesProvider: FC<IWorkspacesProviderProps> = ({
       // setWorkspace(next)
       // localStorage.setItem('workspace', JSON.stringify(next))
       handleUpdateWorkspace(next);
+      localStorage.removeItem("workflowEdges");
+      localStorage.removeItem("workflowNodes");
+      localStorage.removeItem("workflowPieces");
+      localStorage.removeItem("workflowPiecesData");
+      localStorage.removeItem("workflowSettingsData");
     },
     [workspaces, handleUpdateWorkspace],
   );

frontend/src/features/workflowEditor/context/workflowsEditor.tsx

@@ -232,7 +232,7 @@ const WorkflowsEditorProvider: FC<{ children?: React.ReactNode }> = ({
             return acc;
           }
 
-          if (!value.fromUpstream && !value.value) {
+          if (!value.fromUpstream && typeof value.value === 'string' && !value.value) {
             return acc;
           }
           if (

pyproject.toml

@@ -69,7 +69,8 @@ cli = [
     "docker>=7.0.0",
     "kubernetes==23.6.0",
     "bottle==0.12.25",
-    "requests==2.31.0"
+    "requests==2.31.0",
+    "pytest==8.2.2"
 ]
 airflow = [
     "apache-airflow==2.7.2",

rest/services/workflow_service.py

@@ -78,11 +78,11 @@ def create_workflow(
         new_workflow = Workflow(
             name=body.workflow.name,
             uuid_name=workflow_id,
-            created_at=datetime.utcnow(),
+            created_at=datetime.now(),
             schema=body.forageSchema,
             ui_schema=body.ui_schema.model_dump(),
             created_by=auth_context.user_id,
-            last_changed_at=datetime.utcnow(),
+            last_changed_at=datetime.now(),
             start_date=body.workflow.start_date,
             end_date=body.workflow.end_date,
             schedule=body.workflow.schedule,

src/domino/cli/cli.py

@@ -8,6 +8,8 @@
 import ast
 import domino
 
+from domino.cli.utils.constants import COLOR_PALETTE
+
 console = Console()
 
 # Ref: https://patorjk.com/software/taag/
@@ -265,6 +267,40 @@ def generate_random_repo_name():
     return f"new_repository_{str(uuid.uuid4())[0:8]}"
 
 
+@click.command()
+@click.option(
+    '--name',
+    default="ExamplePiece",
+    help='Piece name'
+)
+@click.option(
+    '--repository-path',
+    default=None,
+    help='Path of piece repository.'
+)
+def cli_create_piece(name: str, repository_path: str = None):
+    """Create piece."""
+    try:
+        if repository_path is not None:
+            pieces_repository.create_piece(name, f"{repository_path}/pieces")
+        elif not (Path.cwd() / "pieces").is_dir():
+            # might be called inside the pieces directory
+            if Path.cwd().name == "pieces":
+                pieces_repository.create_piece(name, str(Path.cwd()))
+            else:
+                raise FileNotFoundError("No pieces directory found.")
+        else:
+            pieces_repository.create_piece(name, f"{Path.cwd()}/pieces")
+    except FileNotFoundError as err:
+        console.print(err, style=f"bold {COLOR_PALETTE.get('error')}")
+
+@click.group()
+def cli_pieces():
+    """Manage pieces in a repository."""
+    pass
+
+cli_pieces.add_command(cli_create_piece, name="create")
+
 @click.command()
 @click.option(
     '--name',
@@ -368,19 +404,18 @@ def cli_delete_release(tag_name: str):
 
 @click.group()
 @click.pass_context
-def cli_piece(ctx):
+def cli_piece_repository(ctx):
     """Pieces repository actions"""
     if ctx.invoked_subcommand == "organize":
         console.print(f"Organizing Pieces Repository at: {Path('.').resolve()}")
     elif ctx.invoked_subcommand == "create":
         pass
 
 
-cli_piece.add_command(cli_organize_pieces_repository, name="organize")
-cli_piece.add_command(cli_create_piece_repository, name="create")
-cli_piece.add_command(cli_create_release, name="release")
-cli_piece.add_command(cli_delete_release, name="delete-release")
-cli_piece.add_command(cli_publish_images, name="publish-images")
+cli_piece_repository.add_command(cli_organize_pieces_repository, name="organize")
+cli_piece_repository.add_command(cli_create_release, name="release")
+cli_piece_repository.add_command(cli_delete_release, name="delete-release")
+cli_piece_repository.add_command(cli_publish_images, name="publish-images")
 
 
 ###############################################################################
@@ -418,7 +453,8 @@ def cli(ctx):
 
 
 cli.add_command(cli_platform, name="platform")
-cli.add_command(cli_piece, name="piece")
+cli.add_command(cli_piece_repository, name="piece-repository")
+cli.add_command(cli_pieces, name="pieces")
 cli.add_command(cli_run_piece_k8s, name="run-piece-k8s")
 cli.add_command(cli_run_piece_docker, name='run-piece-docker')
 

src/domino/cli/tests/test_create_piece.py

@@ -0,0 +1,109 @@
+from pathlib import Path
+
+import pytest
+from click.testing import CliRunner
+from domino.cli import cli
+
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+def test_create_piece_success_in_pieces_dir_without_repository_path(
+    runner, tmpdir, monkeypatch
+):
+    pieces_path = str(Path(tmpdir.mkdir("pieces")))
+    piece_name = "TestPiece"
+    monkeypatch.chdir(pieces_path)
+    result = runner.invoke(cli.cli_create_piece, ["--name", f"{piece_name}"])
+    assert result.exit_code == 0
+
+
+def test_create_piece_success_in_repository_dir_without_repository_path(
+    runner, tmpdir, monkeypatch
+):
+    repository_path = Path(tmpdir.mkdir("repo"))
+    piece_name = "TestPiece"
+    monkeypatch.chdir(repository_path)
+    result = runner.invoke(cli.cli_create_piece, ["--name", f"{piece_name}"])
+    assert result.exit_code == 0
+
+
+def test_create_piece_success_with_repository_path(runner, tmpdir):
+    repository_path = Path(tmpdir.mkdir("repo"))
+    tmpdir.mkdir("repo/pieces")
+    piece_name = "TestPiece"
+    result = runner.invoke(
+        cli.cli_create_piece,
+        ["--name", f"{piece_name}", "--repository-path", f"{repository_path}"],
+    )
+    assert result.exit_code == 0
+
+
+def test_create_piece_success_in_pieces_dir_without_args(runner, tmpdir):
+    tmpdir.mkdir("repo")
+    tmpdir.mkdir("repo/pieces")
+    result = runner.invoke(cli.cli_create_piece)
+    assert result.exit_code == 0
+
+
+@pytest.mark.parametrize(
+    "piece_name",
+    [
+        "1TestPiece",
+        "Test",
+        "TestPiec",
+        "Test Piece",
+        "Testpiece",
+        "TESTPIECE",
+        "",
+        " ",
+        ".",
+    ],
+)
+def test_create_piece_fail_invalid_piece_name(runner, tmpdir, piece_name):
+    repository_path = (Path(tmpdir.mkdir("repo")),)
+    result = runner.invoke(
+        cli.cli_create_piece,
+        ["--name", f"{piece_name}", "--repository-path", f"{repository_path}"],
+    )
+    if len(piece_name) < 1:
+        assert "Piece name must have at least one character." in result.output
+    else:
+        assert (
+            f"Validation Error: {piece_name} is not a valid piece name."
+            in result.output
+        )
+
+
+def test_create_piece_already_exists(runner, tmpdir):
+    repository_path = Path(tmpdir.mkdir("repo"))
+    tmpdir.mkdir("repo/pieces")
+    piece_name = "TestPiece"
+    runner.invoke(
+        cli.cli_create_piece,
+        ["--name", f"{piece_name}", "--repository-path", f"{repository_path}"],
+    )
+    result = runner.invoke(
+        cli.cli_create_piece,
+        ["--name", f"{piece_name}", "--repository-path", f"{repository_path}"],
+    )
+    assert f"{piece_name} is already exists" in result.output
+
+
+def test_create_piece_invalid_pieces_path(runner, tmpdir):
+    repository_path = Path(tmpdir.mkdir("repo"))
+    piece_name = "TestPiece"
+    result = runner.invoke(
+        cli.cli_create_piece,
+        ["--name", f"{piece_name}", "--repository-path", f"{repository_path}"],
+    )
+    assert f"{repository_path / 'pieces'} is not a valid repository path."
+
+
+def test_create_piece_pieces_directory_not_exists(runner, tmpdir, monkeypatch):
+    repository_path = Path(tmpdir.mkdir("repo"))
+    monkeypatch.chdir(repository_path)
+    result = runner.invoke(cli.cli_create_piece, ["--name", "TestPiece"])
+    assert "No pieces directory found." in result.output

src/domino/cli/utils/docker-compose-without-database.yaml

@@ -1,7 +1,6 @@
 # Basic Airflow cluster configuration for CeleryExecutor with Redis and PostgreSQL.
 # WARNING: This configuration is for local development. Do not use it in a production deployment.
 ---
-version: '3.8'
 x-airflow-common:
   &airflow-common
   image: apache/airflow:2.7.2-python3.9

src/domino/cli/utils/docker-compose.yaml

@@ -1,7 +1,6 @@
 # Basic Airflow cluster configuration for CeleryExecutor with Redis and PostgreSQL.
 # WARNING: This configuration is for local development. Do not use it in a production deployment.
 ---
-version: '3.8'
 x-airflow-common:
   &airflow-common
   image: apache/airflow:2.7.2-python3.9

src/domino/cli/utils/pieces_repository.py

@@ -16,6 +16,8 @@
 from domino.cli.utils.constants import COLOR_PALETTE
 from domino.client.github_rest_client import GithubRestClient
 from domino.utils import dict_deep_update
+from domino.exceptions.exceptions import ValidationError
+from domino.cli.utils import templates
 
 
 console = Console()
@@ -215,6 +217,52 @@ def validate_pieces_folders() -> None:
     if len(missing_dependencies_errors) > 0:
         raise Exception("\n" + "\n".join(missing_dependencies_errors))
 
+def _validate_piece_name(name: str):
+    """
+    Validate given piece name.
+    """
+    if len(name) == 0:
+        raise ValidationError(f"Piece name must have at least one character.")
+    regex = r'^[A-Za-z_][A-Za-z0-9_]*Piece$'
+    pattern = re.compile(regex)
+    if not pattern.match(name):
+        raise ValidationError(f"{name} is not a valid piece name. Piece name must be valid Python class name and must end with 'Piece'.")
+
+def create_piece(name: str, piece_repository: str):
+    """
+    Create a new piece directory with necessary files.
+    """
+    try:
+        _validate_piece_name(name)
+        piece_dir = os.path.join(piece_repository, name)
+        os.mkdir(piece_dir)
+
+        with open(f"{piece_dir}/piece.py", "x") as f:
+            piece = templates.piece_function(name)
+            f.write(piece)
+
+        with open(f"{piece_dir}/models.py", "x") as f:
+            models = templates.piece_models(name)
+            f.write(models)
+
+        with open(f"{piece_dir}/test_{name}.py", "x") as f:
+            test = templates.piece_test(name)
+            f.write(test)
+
+        with open(f"{piece_dir}/metadata.json", "x") as f:
+            metadata = templates.piece_metadata(name)
+            json.dump(metadata, f, indent = 4)
+
+        console.print(f"{name} is created in {piece_repository}.", style=f"bold {COLOR_PALETTE.get('success')}")
+    except ValidationError as err:
+        console.print(f"{err}", style=f"bold {COLOR_PALETTE.get('error')}")
+    except OSError as err: # todo: create a wrapper for this
+        if err.errno == 17:
+            console.print(f"{name} is already exists in {piece_repository}.", style=f"bold {COLOR_PALETTE.get('error')}")
+        elif err.errno == 2:
+            console.print(f"{piece_repository} is not a valid repository path.", style=f"bold {COLOR_PALETTE.get('error')}")
+        else:
+            console.print(f"{err}", style=f"bold {COLOR_PALETTE.get('error')}")
 
 def create_pieces_repository(repository_name: str, container_registry: str) -> None:
     """
@@ -512,4 +560,4 @@ def delete_release(tag_name: str):
             time.sleep(5)  # Wait for 5 seconds before checking again
         console.print(f"Deletion error: Release {tag_name} still exists after {timeout} seconds.", style=f"bold {COLOR_PALETTE.get('warning')}")
     else:
-        console.print(f"Release {tag_name} not found. Skipping deletion.", style=f"bold {COLOR_PALETTE.get('warning')}")
+        console.print(f"Release {tag_name} not found. Skipping deletion.", style=f"bold {COLOR_PALETTE.get('warning')}")

src/domino/cli/utils/platform.py

@@ -330,6 +330,14 @@ def create_platform(install_airflow: bool = True, use_gpu: bool = False) -> None
     )
     airflow_ssh_config_parsed = AsLiteral(yaml.dump(airflow_ssh_config))
 
+    extra_env = [
+        dict(
+            name='AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL',
+            value="10"
+        )
+    ]
+    extra_env_parsed = AsLiteral(yaml.dump(extra_env))
+
     workers_extra_volumes = []
     workers_extra_volumes_mounts = []
     workers = {}
@@ -382,6 +390,7 @@ def create_platform(install_airflow: bool = True, use_gpu: bool = False) -> None
                 "data": airflow_ssh_config_parsed
             }
         },
+        "extraEnv": extra_env_parsed,
         "config": {
             "api": {
                 "auth_backends": "airflow.api.auth.backend.basic_auth"
@@ -390,7 +399,7 @@ def create_platform(install_airflow: bool = True, use_gpu: bool = False) -> None
         "dags": {
             "gitSync": {
                 "enabled": True,
-                "wait": 60,
+                "wait": 10,
                 "repo": f"ssh://[email protected]/{platform_config['github']['DOMINO_GITHUB_WORKFLOWS_REPOSITORY']}.git",
                 "branch": "main",
                 "subPath": "workflows",