Always pull images in local runner (#279)

Fixes #259 

This PR:
- Instructs the local runner to always pull images. This makes sure that
moving tags like `dev` or `latest` are up to date before running the
pipeline. As far as I can tell from my testing, this doesn't cause a
noticeable delay when the image was not updated.
- Adds the pipeline name to the docker compose yaml and removes orphan
containers. Containers are namespaced by the pipeline name, which
defaults to the cwd if not specified. So I set the name explicitly so we
don't accidentally remove containers from another pipeline.

fondant/compiler.py

@@ -178,4 +178,8 @@ def _generate_spec(self, pipeline: Pipeline, extra_volumes: list) -> dict:
                 services[safe_component_name][
                     "image"
                 ] = component_op.component_spec.image
-        return {"version": "3.8", "services": services}
+        return {
+            "name": pipeline.name,
+            "version": "3.8",
+            "services": services,
+        }

fondant/pipeline.py

@@ -1,6 +1,7 @@
 """This module defines classes to represent a Fondant Pipeline."""
 import json
 import logging
+import re
 import typing as t
 from collections import OrderedDict
 from pathlib import Path
@@ -145,7 +146,7 @@ def __init__(
             pipeline_description: Optional description of the pipeline.
         """
         self.base_path = base_path
-        self.name = pipeline_name
+        self.name = self._validate_pipeline_name(pipeline_name)
         self.description = pipeline_description
         self.package_path = f"{pipeline_name}.tgz"
         self._graph: t.OrderedDict[str, t.Any] = OrderedDict()
@@ -218,6 +219,14 @@ def depth_first_traversal(node: str):
 
         self._graph = OrderedDict((node, self._graph[node]) for node in sorted_graph)
 
+    @staticmethod
+    def _validate_pipeline_name(pipeline_name: str) -> str:
+        pattern = r"^[a-z0-9][a-z0-9_-]*$"
+        if not re.match(pattern, pipeline_name):
+            msg = f"The pipeline name violates the pattern {pattern}"
+            raise InvalidPipelineDefinition(msg)
+        return pipeline_name
+
     def _validate_pipeline_definition(self, run_id: str):
         """
         Validates the pipeline definition by ensuring that the consumed and produced subsets and

fondant/runner.py

@@ -13,6 +13,16 @@ def run(self, *args, **kwargs):
 class DockerRunner(Runner):
     def run(cls, input_spec: str, *args, **kwargs):
         """Run a docker-compose spec."""
-        cmd = ["docker", "compose", "-f", input_spec, "up", "--build"]
+        cmd = [
+            "docker",
+            "compose",
+            "-f",
+            input_spec,
+            "up",
+            "--build",
+            "--pull",
+            "always",
+            "--remove-orphans",
+        ]
 
         subprocess.call(cmd)  # nosec

tests/example_pipelines/compiled_pipeline/example_1/docker-compose.yml

@@ -1,3 +1,4 @@
+name: test_pipeline
 services:
   first_component:
     build: ./tests/example_pipelines/valid_pipeline/example_1

tests/example_pipelines/compiled_pipeline/example_2/docker-compose.yml

@@ -1,3 +1,4 @@
+name: test_pipeline
 services:
   first_component:
     build: ./tests/example_pipelines/valid_pipeline/example_1

tests/test_cli.py

@@ -68,7 +68,17 @@ def test_run_logic(tmp_path_factory):
     with patch("subprocess.call") as mock_call:
         run(args)
         mock_call.assert_called_once_with(
-            ["docker", "compose", "-f", "some/path", "up", "--build"],
+            [
+                "docker",
+                "compose",
+                "-f",
+                "some/path",
+                "up",
+                "--build",
+                "--pull",
+                "always",
+                "--remove-orphans",
+            ],
         )
 
     with patch("subprocess.call") as mock_call, tmp_path_factory.mktemp("temp") as fn:
@@ -87,6 +97,9 @@ def test_run_logic(tmp_path_factory):
                 str(fn / "docker-compose.yml"),
                 "up",
                 "--build",
+                "--pull",
+                "always",
+                "--remove-orphans",
             ],
         )
     args2 = argparse.Namespace(kubeflow=True, local=False, ref="some/path")

tests/test_pipeline.py

@@ -238,3 +238,9 @@ def test_defining_reusable_component_op_with_custom_spec():
     assert load_from_hub_op_default_op.component_spec == load_from_hub_op_default_spec
     assert load_from_hub_op_default_op.component_spec != load_from_hub_op_custom_spec
     assert load_from_hub_op_custom_op.component_spec != load_from_hub_op_default_spec
+
+
+def test_pipeline_name():
+    Pipeline(pipeline_name="valid-name", base_path="base_path")
+    with pytest.raises(InvalidPipelineDefinition, match="The pipeline name violates"):
+        Pipeline(pipeline_name="invalid name", base_path="base_path")

tests/test_runner.py

@@ -8,5 +8,15 @@ def test_docker_runner():
     with patch("subprocess.call") as mock_call:
         DockerRunner().run("some/path")
         mock_call.assert_called_once_with(
-            ["docker", "compose", "-f", "some/path", "up", "--build"],
+            [
+                "docker",
+                "compose",
+                "-f",
+                "some/path",
+                "up",
+                "--build",
+                "--pull",
+                "always",
+                "--remove-orphans",
+            ],
         )