Feature:3973 Extend list pipeline filters

- Filter pipelines by the name or ID of the latest executor user
- Extend client.list_pipelines with new filter option.
- Integration tests with multi user switch

Author: Json-Andriopoulos

src/zenml/client.py

@@ -2410,6 +2410,7 @@ def list_pipelines(
         updated: Optional[Union[datetime, str]] = None,
         name: Optional[str] = None,
         latest_run_status: Optional[str] = None,
+        latest_run_user: Optional[Union[UUID, str]] = None,
         project: Optional[Union[str, UUID]] = None,
         user: Optional[Union[UUID, str]] = None,
         tag: Optional[str] = None,
@@ -2429,6 +2430,8 @@ def list_pipelines(
             name: The name of the pipeline to filter by.
             latest_run_status: Filter by the status of the latest run of a
                 pipeline.
+            latest_run_user: Filter by the name or UUID of the user that
+                executed the latest run.
             project: The project name/ID to filter by.
             user: The name/ID of the user to filter by.
             tag: Tag to filter by.
@@ -2449,6 +2452,7 @@ def list_pipelines(
             updated=updated,
             name=name,
             latest_run_status=latest_run_status,
+            latest_run_user=latest_run_user,
             project=project or self.active_project.id,
             user=user,
             tag=tag,

src/zenml/models/v2/core/pipeline.py

@@ -21,6 +21,7 @@
     Optional,
     Type,
     TypeVar,
+    Union,
 )
 from uuid import UUID
 
@@ -268,6 +269,7 @@ class PipelineFilter(ProjectScopedFilter, TaggableFilter):
         *ProjectScopedFilter.FILTER_EXCLUDE_FIELDS,
         *TaggableFilter.FILTER_EXCLUDE_FIELDS,
         "latest_run_status",
+        "latest_run_user",
     ]
     CLI_EXCLUDE_FIELDS: ClassVar[List[str]] = [
         *ProjectScopedFilter.CLI_EXCLUDE_FIELDS,
@@ -284,6 +286,20 @@ class PipelineFilter(ProjectScopedFilter, TaggableFilter):
         "This will always be applied as an `AND` filter for now.",
     )
 
+    latest_run_user: Optional[Union[UUID, str]] = Field(
+        default=None,
+        description="Filter by the name or id of the last user that executed the pipeline. ",
+    )
+
+    @property
+    def filter_by_latest_execution(self) -> bool:
+        """Property returning whether filtering considers latest pipeline execution.
+
+        Returns:
+            True if latest pipeline execution filters are used (e.g. latest_run_status).
+        """
+        return bool(self.latest_run_user) or bool(self.latest_run_status)
+
     def apply_filter(
         self, query: AnyQuery, table: Type["AnySchema"]
     ) -> AnyQuery:
@@ -300,9 +316,13 @@ def apply_filter(
 
         from sqlmodel import and_, col, func, select
 
-        from zenml.zen_stores.schemas import PipelineRunSchema, PipelineSchema
+        from zenml.zen_stores.schemas import (
+            PipelineRunSchema,
+            PipelineSchema,
+            UserSchema,
+        )
 
-        if self.latest_run_status:
+        if self.filter_by_latest_execution:
             latest_pipeline_run_subquery = (
                 select(
                     PipelineRunSchema.pipeline_id,
@@ -313,28 +333,39 @@ def apply_filter(
                 .subquery()
             )
 
-            query = (
-                query.join(
-                    PipelineRunSchema,
-                    PipelineSchema.id == PipelineRunSchema.pipeline_id,
+            query = query.join(
+                PipelineRunSchema,
+                PipelineSchema.id == PipelineRunSchema.pipeline_id,
+            ).join(
+                latest_pipeline_run_subquery,
+                and_(
+                    PipelineRunSchema.pipeline_id
+                    == latest_pipeline_run_subquery.c.pipeline_id,
+                    PipelineRunSchema.created
+                    == latest_pipeline_run_subquery.c.created,
+                ),
+            )
+
+            if self.latest_run_user:
+                query = query.join(
+                    UserSchema, UserSchema.id == PipelineRunSchema.user_id
                 )
-                .join(
-                    latest_pipeline_run_subquery,
-                    and_(
-                        PipelineRunSchema.pipeline_id
-                        == latest_pipeline_run_subquery.c.pipeline_id,
-                        PipelineRunSchema.created
-                        == latest_pipeline_run_subquery.c.created,
-                    ),
+
+                query = query.where(
+                    self.generate_name_or_id_query_conditions(
+                        value=self.latest_run_user,
+                        table=UserSchema,
+                    )
                 )
-                .where(
+
+            if self.latest_run_status:
+                query = query.where(
                     self.generate_custom_query_conditions_for_column(
                         value=self.latest_run_status,
                         table=PipelineRunSchema,
                         column="status",
                     )
                 )
-            )
 
         return query
 

tests/integration/functional/pipelines/test_multi_user_crud.py

@@ -0,0 +1,109 @@
+import uuid
+
+import pytest
+from tests.integration.functional.zen_stores.utils import LoginContext
+
+from zenml import pipeline, step
+from zenml.client import Client
+from zenml.config.global_config import GlobalConfiguration
+from zenml.enums import StoreType
+from zenml.models.v2.core.user import UserResponse
+
+
+@step
+def greet_step(name: str) -> str:
+    return f"Welcome {name}"
+
+
+@step
+def crush_step():
+    raise ValueError("Oh noooo...")
+
+
+@pipeline
+def pipeline_that_completes():
+    greet_step("test")
+
+
+@pipeline()
+def pipeline_that_crushes():
+    crush_step()
+
+
+def submit_pipelines_programmatically(client: Client):
+    pipeline_that_completes()
+
+    try:
+        pipeline_that_crushes()
+    except RuntimeError:
+        pass
+
+
+def check_pipelines(client: Client, user_name: str):
+    pipelines = client.list_pipelines(
+        latest_run_status="completed", name="pipeline_that_completes"
+    )
+
+    assert pipelines.size == 1
+
+    pipelines = client.list_pipelines(
+        latest_run_status="failed", name="pipeline_that_crushes"
+    )
+
+    assert pipelines.size == 1
+
+    pipelines = client.list_pipelines(
+        name="pipeline_that_completes", latest_run_user=user_name
+    )
+
+    assert pipelines.size == 1
+
+    pipelines = client.list_pipelines(
+        name="pipeline_that_crushes", latest_run_user=user_name
+    )
+
+    assert pipelines.size == 1
+
+
+def check_all_user_id_alternatives_work(client: Client, user: UserResponse):
+    pipelines = client.list_pipelines(
+        name="pipeline_that_completes", latest_run_user=user.id
+    )
+
+    assert pipelines.size == 1
+
+    pipelines = client.list_pipelines(
+        name="pipeline_that_completes", latest_run_user=str(user.id)
+    )
+
+    assert pipelines.size == 1
+
+    pipelines = client.list_pipelines(
+        name="pipeline_that_completes", latest_run_user=user.name
+    )
+
+    assert pipelines.size == 1
+
+
+def test_multi_user_pipeline_executions():
+    if GlobalConfiguration().zen_store.config.type != StoreType.REST:
+        pytest.skip("Multi-user pipelines are supported only over REST.")
+
+    client = Client()
+
+    submit_pipelines_programmatically(client)
+
+    check_pipelines(client, user_name="default")
+
+    user = client.create_user(
+        name=f"tester_{uuid.uuid4()}", password="1234", is_admin=True
+    )
+
+    with LoginContext(user_name=user.name, password="1234"):
+        new_client = Client()
+
+        submit_pipelines_programmatically(new_client)
+
+        check_pipelines(new_client, user.name)
+
+        check_all_user_id_alternatives_work(new_client, user=user)

tests/unit/models/test_pipeline_models.py

@@ -18,7 +18,7 @@
 
 from zenml.config.pipeline_spec import PipelineSpec
 from zenml.constants import STR_FIELD_MAX_LENGTH, TEXT_FIELD_MAX_LENGTH
-from zenml.models import PipelineRequest
+from zenml.models import PipelineFilter, PipelineRequest
 
 
 def test_pipeline_request_model_fails_with_long_name():
@@ -41,3 +41,26 @@ def test_pipeline_request_model_fails_with_long_docstring():
             docstring=long_docstring,
             spec=PipelineSpec(steps=[]),
         )
+
+
+def test_pipeline_filter_by_latest_execution():
+    f = PipelineFilter()
+
+    assert not f.filter_by_latest_execution
+
+    f = PipelineFilter(latest_run_status="completed")
+
+    assert f.filter_by_latest_execution
+
+    f = PipelineFilter(latest_run_user="test")
+
+    assert f.filter_by_latest_execution
+
+    # make sure latest pipeline run associated fields are not propagated to base table filters.
+
+    latest_run_fields = [
+        "latest_run_status",
+        "latest_run_user",
+    ]
+
+    assert all(field in f.FILTER_EXCLUDE_FIELDS for field in latest_run_fields)