Merge branch 'feature/improve-cli-tables' into experiment/try-partial-selective-ci-tests

Author: safoinme

.github/workflows/claude-code-review.yml

@@ -30,7 +30,7 @@ jobs:
       needs.check-team-member.outputs.is-team-member == 'true' &&
       github.actor != 'github-actions[bot]' &&
       github.event.issue.pull_request &&
-      contains(github.event.comment.body, '/claude /full-review')
+      contains(github.event.comment.body, '@claude /full-review')
     runs-on: ubuntu-latest
     timeout-minutes: 10
     concurrency:

.github/workflows/claude.yml

@@ -36,10 +36,10 @@ jobs:
       needs.check-team-member.outputs.is-team-member == 'true' &&
       github.actor != 'github-actions[bot]' &&
       (
-        (github.event_name == 'issue_comment' && contains(github.event.comment.body, '/claude') && !contains(github.event.comment.body, '/claude /full-review')) ||
-        (github.event_name == 'pull_request_review_comment' && contains(github.event.comment.body, '/claude') && !contains(github.event.comment.body, '/claude /full-review')) ||
-        (github.event_name == 'pull_request_review' && contains(github.event.review.body, '/claude') && !contains(github.event.review.body, '/claude /full-review')) ||
-        (github.event_name == 'issues' && (contains(github.event.issue.body, '/claude') || contains(github.event.issue.title, '/claude')))
+        (github.event_name == 'issue_comment' && contains(github.event.comment.body, '@claude') && !contains(github.event.comment.body, '@claude /full-review')) ||
+        (github.event_name == 'pull_request_review_comment' && contains(github.event.comment.body, '@claude') && !contains(github.event.comment.body, '@claude /full-review')) ||
+        (github.event_name == 'pull_request_review' && contains(github.event.review.body, '@claude') && !contains(github.event.review.body, '@claude /full-review')) ||
+        (github.event_name == 'issues' && (contains(github.event.issue.body, '@claude') || contains(github.event.issue.title, '@claude')))
       )
     runs-on: ubuntu-latest
     timeout-minutes: 10

docs/book/how-to/steps-pipelines/dynamic_pipelines.md

@@ -5,7 +5,7 @@ description: Write dynamic pipelines
 # Dynamic Pipelines (Experimental)
 
 {% hint style="warning" %}
-**Experimental Feature**: Dynamic pipelines are currently an experimental feature. There are known issues and limitations, and the interface is subject to change. This feature is only supported by the `local` and `kubernetes` orchestrators. If you encounter any issues or have feedback, please let us know at [https://github.com/zenml-io/zenml/issues](https://github.com/zenml-io/zenml/issues).
+**Experimental Feature**: Dynamic pipelines are currently an experimental feature. There are known issues and limitations, and the interface is subject to change. This feature is only supported by the `local`, `kubernetes`, `sagemaker` and `vertex` orchestrators. If you encounter any issues or have feedback, please let us know at [https://github.com/zenml-io/zenml/issues](https://github.com/zenml-io/zenml/issues).
 {% endhint %}
 
 {% hint style="info" %}
@@ -180,6 +180,39 @@ def unmapped_example():
     consumer.map(a=a, b=unmapped(b))
 ```
 
+#### Unpacking mapped outputs
+
+If a mapped step returns multiple outputs, you can split them into separate lists (one per output) using `unpack()`. This returns a tuple of lists of artifact futures, aligned by mapped invocation.
+
+```python
+from zenml import pipeline, step
+
+@step
+def create_int_list() -> list[int]:
+    return [1, 2]
+
+@step
+def compute(a: int) -> tuple[int, int]:
+    return a * 2, a * 3
+
+@pipeline(dynamic=True)
+def map_pipeline():
+    ints = create_int_list()
+    results = compute.map(a=ints)  # Map over [1, 2]
+
+    # Unpack per-output across all mapped invocations
+    double, triple = results.unpack()
+
+    # Each element is an ArtifactFuture; load to get concrete values
+    doubles = [f.load() for f in double]  # [2, 4]
+    triples = [f.load() for f in triple]  # [3, 6]
+```
+
+Notes:
+- `results` is a future that refers to all outputs of all steps, and `unpack()` works for both `.map(...)` and `.product(...)`.
+- Each list contains future objects that refer to a single artifact.
+
+
 ### Parallel Step Execution
 
 Dynamic pipelines support true parallel execution using `step.submit()`. This method returns a `StepRunFuture` that you can use to wait for results or pass to downstream steps:
@@ -276,26 +309,11 @@ When running multiple steps concurrently using `step.submit()`, a failure in one
 Dynamic pipelines are currently only supported by:
 - `local` orchestrator
 - `kubernetes` orchestrator
+- `sagemaker` orchestrator
+- `vertex` orchestrator
 
 Other orchestrators will raise an error if you try to run a dynamic pipeline with them.
 
-### Remote Execution Requirement
-
-When running dynamic pipelines remotely (e.g., with the `kubernetes` orchestrator), you **must** include `depends_on` for at least one step in your pipeline definition. This is currently required due to a bug in remote execution.
-
-{% hint style="warning" %}
-**Required for Remote Execution**: Without `depends_on`, remote execution will fail. This requirement does not apply when running locally with the `local` orchestrator.
-{% endhint %}
-
-For example:
-
-```python
-@pipeline(dynamic=True, depends_on=[some_step])
-def dynamic_pipeline():
-    some_step()
-    # ... rest of your pipeline
-```
-
 ### Artifact Loading
 
 When you call `.load()` on an artifact in a dynamic pipeline, it synchronously loads the data. For large artifacts or when you want to maintain parallelism, consider passing the step outputs (future or artifact) directly to downstream steps instead of loading them.

src/zenml/cli/utils.py

@@ -3215,7 +3215,7 @@ def handle_output(
             logger.warning("Failed to write clean output: %s", err)
             print(cli_output)
 
-    if page:
+    if page and output_format == "table":
         print_page_info(page)
 
 
@@ -3321,16 +3321,27 @@ def prepare_output(
 def _syntax_highlight(content: str, lexer: str) -> str:
     """Apply syntax highlighting to content if colors are enabled.
 
+    Syntax highlighting is only applied when output goes to an interactive
+    terminal (TTY). When output is redirected to a file or piped to another
+    program, plain text is returned to ensure machine-readable output.
+
     Args:
         content: The text content to highlight
         lexer: The lexer to use (e.g., "json", "yaml")
 
     Returns:
-        Syntax-highlighted string if colors enabled, otherwise original content
+        Syntax-highlighted string if colors enabled and output is a TTY,
+        otherwise the original content unchanged.
     """
     if os.getenv("NO_COLOR"):
         return content
 
+    # Import here to avoid circular imports at module load time
+    from zenml_cli import is_terminal_output
+
+    if not is_terminal_output():
+        return content
+
     syntax = Syntax(
         content, lexer, theme="ansi_dark", background_color="default"
     )

src/zenml/config/docker_settings.py

@@ -55,20 +55,129 @@ class PythonPackageInstaller(Enum):
     UV = "uv"
 
 
+# (docker_sdk_argument, attribute_name)
+BUILD_OPTION_CONVERSIONS = [
+    ("buildargs", "build_args"),
+    ("cachefrom", "cache_from"),
+    ("nocache", "no_cache"),
+    ("shmsize", "shm_size"),
+]
+
+
+class DockerBuildOptions(BaseModel):
+    """Docker build options.
+
+    This class only specifies a subset of the options which require explicit
+    conversion as they require different names in the Docker CLI and Python SDK.
+    However, you can still specify any other options as extra model attributes
+    which will be passed unmodified to the build method of the image builder.
+    """
+
+    pull: Optional[bool] = None
+    rm: Optional[bool] = None
+    no_cache: Optional[bool] = None
+    shm_size: Optional[int] = None
+    labels: Optional[Dict[str, Any]] = None
+    build_args: Optional[Dict[str, Any]] = None
+    cache_from: Optional[List[str]] = None
+
+    model_config = ConfigDict(extra="allow")
+
+    @model_validator(mode="before")
+    @classmethod
+    @before_validator_handler
+    def _migrate_sdk_arguments(cls, data: Dict[str, Any]) -> Dict[str, Any]:
+        """Migrate Docker SDK arguments to attributes.
+
+        Args:
+            data: The model data.
+
+        Returns:
+            The migrated data.
+        """
+        for sdk_argument, attribute_name in BUILD_OPTION_CONVERSIONS:
+            if sdk_argument in data and attribute_name not in data:
+                data[attribute_name] = data.pop(sdk_argument)
+
+        return data
+
+    def to_docker_cli_options(self) -> List[str]:
+        """Convert the build options to a list of Docker CLI options.
+
+        https://docs.docker.com/reference/cli/docker/buildx/build/#options
+
+        Returns:
+            A list of Docker CLI options.
+        """
+        options = []
+        if self.pull:
+            options.append("--pull")
+        if self.rm:
+            options.append("--rm")
+        if self.no_cache:
+            options.append("--no-cache")
+        if self.shm_size:
+            options.extend(["--shm-size", str(self.shm_size)])
+        if self.labels:
+            for key, value in self.labels.items():
+                options.extend(["--label", f"{key}={value}"])
+        if self.build_args:
+            for key, value in self.build_args.items():
+                options.extend(["--build-arg", f"{key}={value}"])
+        if self.cache_from:
+            for value in self.cache_from:
+                options.extend(["--cache-from", value])
+
+        if self.model_extra:
+            for key, value in self.model_extra.items():
+                option = f"--{key.replace('_', '-')}"
+                if isinstance(value, Dict):
+                    for key, value in value.items():
+                        options.extend([option, f"{key}={value}"])
+                elif isinstance(value, List):
+                    for val in value:
+                        options.extend([option, str(val)])
+                elif value in (True, None):
+                    options.extend([option])
+                elif value is False:
+                    pass
+                else:
+                    options.extend([option, str(value)])
+
+        return options
+
+    def to_docker_python_sdk_options(self) -> Dict[str, Any]:
+        """Get the build options as a dictionary of Docker Python SDK options.
+
+        https://docker-py.readthedocs.io/en/stable/images.html#docker.models.images.ImageCollection.build
+
+        Returns:
+            A dictionary of Docker Python SDK options.
+        """
+        options = self.model_dump(exclude_unset=True)
+        for sdk_argument, attribute_name in BUILD_OPTION_CONVERSIONS:
+            if attribute_name in options:
+                options[sdk_argument] = options.pop(attribute_name)
+
+        return options
+
+
 class DockerBuildConfig(BaseModel):
     """Configuration for a Docker build.
 
     Attributes:
-        build_options: Additional options that will be passed unmodified to the
-            Docker build call when building an image. You can use this to for
-            example specify build args or a target stage. See
-            https://docker-py.readthedocs.io/en/stable/images.html#docker.models.images.ImageCollection.build
-            for a full list of available options.
+        build_options: Additional options that will be passed when building an
+            image. Depending on the image builder that is used, different
+            options are available.
+            For image builders that use the Docker CLI:
+            - https://docs.docker.com/reference/cli/docker/buildx/build/#options
+            For image builders that use the Docker Python SDK:
+            - https://docker-py.readthedocs.io/en/stable/images.html#docker.models.images.ImageCollection.build
         dockerignore: Path to a dockerignore file to use when building the
             Docker image.
     """
 
-    build_options: Dict[str, Any] = {}
+    build_options: Optional[DockerBuildOptions] = None
     dockerignore: Optional[str] = None
 
 

src/zenml/execution/pipeline/dynamic/outputs.py

@@ -14,7 +14,7 @@
 """Dynamic pipeline execution outputs."""
 
 from concurrent.futures import Future
-from typing import Any, List, Optional, Tuple, Union, overload
+from typing import Any, Iterator, List, Optional, Tuple, Union, overload
 
 from zenml.logger import get_logger
 from zenml.models import ArtifactVersionResponse
@@ -270,4 +270,92 @@ def __len__(self) -> int:
         return len(self._output_keys)
 
 
-StepRunFuture = Union[ArtifactFuture, StepRunOutputsFuture]
+class MapResultsFuture:
+    """Future that represents the results of a `step.map/product(...)` call."""
+
+    def __init__(self, futures: List[StepRunOutputsFuture]) -> None:
+        """Initialize the map results future.
+
+        Args:
+            futures: The step run futures.
+        """
+        self.futures = futures
+
+    def result(self) -> List[StepRunOutputs]:
+        """Get the step run outputs this future represents.
+
+        Returns:
+            The step run outputs.
+        """
+        return [future.result() for future in self.futures]
+
+    def unpack(self) -> Tuple[List[ArtifactFuture], ...]:
+        """Unpack the map results future.
+
+        This method can be used to get lists of artifact futures that represent
+        the outputs of all the step runs that are part of this map result.
+
+        Example:
+        ```python
+        from zenml import pipeline, step
+
+        @step
+        def create_int_list() -> list[int]:
+            return [1, 2]
+
+        @step
+        def do_something(a: int) -> Tuple[int, int]:
+            return a * 2, a * 3
+
+        @pipeline
+        def map_pipeline():
+            int_list = create_int_list()
+            results = do_something.map(a=int_list)
+            double, triple = results.unpack()
+
+            # [future.load() for future in double] will return [2, 4]
+            # [future.load() for future in triple] will return [3, 6]
+        ```
+
+        Returns:
+            The unpacked map results.
+        """
+        return tuple(map(list, zip(*self.futures)))
+
+    @overload
+    def __getitem__(self, key: int) -> StepRunOutputsFuture: ...
+
+    @overload
+    def __getitem__(self, key: slice) -> List[StepRunOutputsFuture]: ...
+
+    def __getitem__(
+        self, key: Union[int, slice]
+    ) -> Union[StepRunOutputsFuture, List[StepRunOutputsFuture]]:
+        """Get a step run future.
+
+        Args:
+            key: The index or slice of the step run futures.
+
+        Returns:
+            The step run futures.
+        """
+        return self.futures[key]
+
+    def __iter__(self) -> Iterator[StepRunOutputsFuture]:
+        """Iterate over the step run futures.
+
+        Yields:
+            The step run futures.
+        """
+        yield from self.futures
+
+    def __len__(self) -> int:
+        """Get the number of step run futures.
+
+        Returns:
+            The number of step run futures.
+        """
+        return len(self.futures)
+
+
+StepRunFuture = Union[ArtifactFuture, StepRunOutputsFuture, MapResultsFuture]