feat: add runnable config dependency

Author: cuinixam

src/py_app_dev/core/runnable.py

@@ -29,6 +29,14 @@ def get_inputs(self) -> List[Path]:
     def get_outputs(self) -> List[Path]:
         """Get runnable outputs."""
 
+    def get_config(self) -> Optional[dict[str, str]]:
+        """
+        Get runnable configuration.
+
+        (!) Do NOT put sensitive information in the configuration. It will be stored in a file.
+        """
+        return None
+
 
 class RunInfoStatus(Enum):
     MATCH = (False, "Nothing changed. Previous execution info matches.")
@@ -37,6 +45,7 @@ class RunInfoStatus(Enum):
     FILE_CHANGED = (True, "File has changed.")
     NOTHING_TO_CHECK = (True, "Nothing to be checked. Assume it shall always run.")
     FORCED_RUN = (True, "Forced run. Ignore previous execution info.")
+    CONFIG_CHANGED = (True, "Configuration has changed.")
 
     def __init__(self, should_run: bool, message: str) -> None:
         self.should_run = should_run
@@ -84,6 +93,11 @@ def file_hash_to_str(file_hash: Optional[str]) -> str:
             "outputs": {str(path): file_hash_to_str(self.get_file_hash(path)) for path in runnable.get_outputs()},
         }
 
+        # Only store config if the runnable has a config
+        config = runnable.get_config()
+        if config is not None:
+            file_info["config"] = config
+
         run_info_path = self.get_runnable_run_info_file(runnable)
         run_info_path.parent.mkdir(parents=True, exist_ok=True)
         with run_info_path.open("w") as f:
@@ -103,6 +117,12 @@ def previous_run_info_matches(self, runnable: Runnable) -> RunInfoStatus:
         with run_info_path.open() as f:
             previous_info = json.load(f)
 
+        # Check if configuration has changed
+        current_config = runnable.get_config()
+        if "config" in previous_info:
+            if current_config != previous_info["config"]:
+                return RunInfoStatus.CONFIG_CHANGED
+
         # Check if there is anything to be checked
         if any(len(previous_info[file_type]) for file_type in ["inputs", "outputs"]):
             for file_type in ["inputs", "outputs"]:

tests/test_runnable.py

@@ -1,3 +1,4 @@
+import json
 import os
 from pathlib import Path
 from typing import List, Optional
@@ -8,7 +9,13 @@
 
 
 class MyRunnable(Runnable):
-    def __init__(self, inputs: Optional[List[Path]] = None, outputs: Optional[List[Path]] = None, return_code: int = 0, needs_dependency_management: bool = True):
+    def __init__(
+        self,
+        inputs: Optional[List[Path]] = None,
+        outputs: Optional[List[Path]] = None,
+        return_code: int = 0,
+        needs_dependency_management: bool = True,
+    ) -> None:
         super().__init__(needs_dependency_management=needs_dependency_management)
         self._inputs = inputs if inputs is not None else []
         self._outputs = outputs if outputs is not None else []
@@ -28,20 +35,20 @@ def get_outputs(self) -> List[Path]:
 
 
 @pytest.fixture
-def executor(tmp_path):
+def executor(tmp_path: Path) -> Executor:
     """Fixture for creating an Executor with a cache directory."""
     cache_dir = tmp_path / "cache"
     cache_dir.mkdir()
     return Executor(cache_dir=cache_dir)
 
 
-def test_no_previous_info(executor):
+def test_no_previous_info(executor: Executor) -> None:
     """Test that Executor correctly detects that a runnable has not been executed before."""
     runnable = MyRunnable()
     assert executor.previous_run_info_matches(runnable) == RunInfoStatus.NO_INFO
 
 
-def test_previous_info_matches(executor, tmp_path):
+def test_previous_info_matches(executor: Executor, tmp_path: Path) -> None:
     """Test that Executor correctly skips execution when previous info matches."""
     input_path = tmp_path / "input.txt"
     output_path = tmp_path / "output.txt"
@@ -54,7 +61,7 @@ def test_previous_info_matches(executor, tmp_path):
     assert new_executor.previous_run_info_matches(runnable) == RunInfoStatus.FORCED_RUN
 
 
-def test_file_changed(executor, tmp_path):
+def test_file_changed(executor: Executor, tmp_path: Path) -> None:
     """Test that Executor correctly detects when a file has changed."""
     input_path = tmp_path / "input.txt"
     input_path.write_text("input")
@@ -64,7 +71,7 @@ def test_file_changed(executor, tmp_path):
     assert executor.previous_run_info_matches(runnable) == RunInfoStatus.FILE_CHANGED
 
 
-def test_file_removed(executor, tmp_path):
+def test_file_removed(executor: Executor, tmp_path: Path) -> None:
     """Test that Executor correctly detects when a file has been removed."""
     output_path = tmp_path / "output.txt"
     output_path.write_text("output")
@@ -74,7 +81,7 @@ def test_file_removed(executor, tmp_path):
     assert executor.previous_run_info_matches(runnable) == RunInfoStatus.FILE_NOT_FOUND
 
 
-def test_directory_exists(executor, tmp_path):
+def test_directory_exists(executor: Executor, tmp_path: Path) -> None:
     """Test that Executor correctly handles existing directories."""
     input_dir = tmp_path / "input_dir"
     output_dir = tmp_path / "output_dir"
@@ -85,7 +92,7 @@ def test_directory_exists(executor, tmp_path):
     assert executor.previous_run_info_matches(runnable) == RunInfoStatus.MATCH
 
 
-def test_directory_removed(executor, tmp_path):
+def test_directory_removed(executor: Executor, tmp_path: Path) -> None:
     """Test that Executor correctly detects when a directory has been removed."""
     input_dir = tmp_path / "input_dir"
     output_dir = tmp_path / "output_dir"
@@ -97,7 +104,7 @@ def test_directory_removed(executor, tmp_path):
     assert executor.previous_run_info_matches(runnable) == RunInfoStatus.FILE_NOT_FOUND
 
 
-def test_mixed_files_and_directories(executor, tmp_path):
+def test_mixed_files_and_directories(executor: Executor, tmp_path: Path) -> None:
     """Test that Executor correctly handles a mix of files and directories."""
     input_file = tmp_path / "input.txt"
     input_dir = tmp_path / "input_dir"
@@ -112,14 +119,14 @@ def test_mixed_files_and_directories(executor, tmp_path):
     assert executor.previous_run_info_matches(runnable) == RunInfoStatus.MATCH
 
 
-def test_no_inputs_and_no_outputs(executor):
+def test_no_inputs_and_no_outputs(executor: Executor) -> None:
     """Test that Executor correctly handles a runnable with no inputs and no outputs."""
     runnable = MyRunnable()
     executor.execute(runnable)
     assert executor.previous_run_info_matches(runnable) == RunInfoStatus.NOTHING_TO_CHECK
 
 
-def test_dry_run(executor):
+def test_dry_run(executor: Executor) -> None:
     """Test that Executor does not execute the run method when dry_run is True."""
     runnable = MyRunnable(return_code=1)
     executor.dry_run = True
@@ -128,9 +135,80 @@ def test_dry_run(executor):
     assert executor.execute(runnable) == 1
 
 
-def test_no_dependency_management(executor):
+def test_no_dependency_management(executor: Executor) -> None:
     """Test that Executor executes runnables without dependency management directly."""
     runnable = MyRunnable(needs_dependency_management=False, return_code=2)
     assert executor.execute(runnable) == 2
     # Ensure it doesn't store or check run info
     assert executor.previous_run_info_matches(runnable) == RunInfoStatus.NO_INFO
+
+
+def test_config_changed(executor: Executor, tmp_path: Path) -> None:
+    """Test that Executor detects when the configuration has changed."""
+    input_path = tmp_path / "input.txt"
+    input_path.write_text("input")
+
+    class ConfigurableRunnable(MyRunnable):
+        def __init__(
+            self,
+            config: dict[str, str],
+            inputs: Optional[List[Path]] = None,
+        ) -> None:
+            super().__init__(inputs=inputs)
+            self._config = config
+
+        def get_config(self) -> Optional[dict[str, str]]:
+            return self._config
+
+    runnable = ConfigurableRunnable(config={"key": "value"}, inputs=[input_path])
+    executor.execute(runnable)
+
+    # Ensure it matches initially
+    assert executor.previous_run_info_matches(runnable) == RunInfoStatus.MATCH
+
+    # Change the configuration
+    runnable = ConfigurableRunnable(config={"key": "new_value"}, inputs=[input_path])
+    assert executor.previous_run_info_matches(runnable) == RunInfoStatus.CONFIG_CHANGED
+
+
+def test_config_stored(executor: Executor, tmp_path: Path) -> None:
+    """Test that Executor stores the configuration alongside inputs and outputs."""
+    input_path = tmp_path / "input.txt"
+    input_path.write_text("input")
+
+    class ConfigurableRunnable(MyRunnable):
+        def __init__(
+            self,
+            config: dict[str, str],
+            inputs: Optional[List[Path]] = None,
+        ) -> None:
+            super().__init__(inputs=inputs)
+            self._config = config
+
+        def get_config(self) -> Optional[dict[str, str]]:
+            return self._config
+
+    config = {"key": "value"}
+    runnable = ConfigurableRunnable(config=config, inputs=[input_path])
+    executor.execute(runnable)
+
+    # Verify the stored run info contains the configuration
+    run_info_path = executor.get_runnable_run_info_file(runnable)
+    with run_info_path.open() as f:
+        run_info = json.load(f)
+    assert run_info["config"] == config
+
+
+def test_config_not_stored_if_none(executor: Executor, tmp_path: Path) -> None:
+    """Test that Executor does not store a config if the runnable has no config."""
+    input_path = tmp_path / "input.txt"
+    input_path.write_text("input")
+
+    runnable = MyRunnable(inputs=[input_path])
+    executor.execute(runnable)
+
+    # Verify the stored run info does not contain a config field
+    run_info_path = executor.get_runnable_run_info_file(runnable)
+    with run_info_path.open() as f:
+        run_info = json.load(f)
+    assert "config" not in run_info