content: #  Copyright (c) ZenML GmbH 2023. All Rights Reserved.
  #
  #  Licensed under the Apache License, Version 2.0 (the "License");
  #  you may not use this file except in compliance with the License.
  #  You may obtain a copy of the License at:
  #
  #       https://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  #  or implied. See the License for the specific language governing
  #  permissions and limitations under the License.
  """Models representing pipeline runs."""
  
  from datetime import datetime
  from typing import (
      TYPE_CHECKING,
      Any,
      ClassVar,
      Dict,
      List,
      Optional,
      Type,
      TypeVar,
      Union,
  )
  from uuid import UUID
  
  from pydantic import ConfigDict, Field
  
  from zenml.config.pipeline_configurations import PipelineConfiguration
  from zenml.constants import STR_FIELD_MAX_LENGTH
  from zenml.enums import ExecutionStatus, PipelineRunTriggeredByType
  from zenml.metadata.metadata_types import MetadataType
  from zenml.models.v2.base.base import BaseUpdate, BaseZenModel
  from zenml.models.v2.base.scoped import (
      ProjectScopedFilter,
      ProjectScopedRequest,
      ProjectScopedResponse,
      ProjectScopedResponseBody,
      ProjectScopedResponseMetadata,
      ProjectScopedResponseResources,
      RunMetadataFilterMixin,
      TaggableFilter,
  )
  from zenml.models.v2.core.logs import LogsRequest
  from zenml.models.v2.core.model_version import ModelVersionResponse
  from zenml.models.v2.core.tag import TagResponse
  from zenml.utils import pagination_utils
  from zenml.utils.tag_utils import Tag
  
  if TYPE_CHECKING:
      from sqlalchemy.sql.elements import ColumnElement
  
      from zenml.models import TriggerExecutionResponse
      from zenml.models.v2.core.artifact_version import ArtifactVersionResponse
      from zenml.models.v2.core.code_reference import CodeReferenceResponse
      from zenml.models.v2.core.curated_visualization import (
          CuratedVisualizationResponse,
      )
      from zenml.models.v2.core.logs import LogsResponse
      from zenml.models.v2.core.pipeline import PipelineResponse
      from zenml.models.v2.core.pipeline_build import (
          PipelineBuildResponse,
      )
      from zenml.models.v2.core.pipeline_snapshot import PipelineSnapshotResponse
      from zenml.models.v2.core.schedule import ScheduleResponse
      from zenml.models.v2.core.stack import StackResponse
      from zenml.models.v2.core.step_run import StepRunResponse
      from zenml.zen_stores.schemas.base_schemas import BaseSchema
  
      AnySchema = TypeVar("AnySchema", bound=BaseSchema)
  
  AnyQuery = TypeVar("AnyQuery", bound=Any)
  
  
  # ------------------ Request Model ------------------
  
  
  class PipelineRunTriggerInfo(BaseZenModel):
      """Trigger information model."""
  
      step_run_id: Optional[UUID] = Field(
          default=None,
          title="The ID of the step run that triggered the pipeline run.",
      )
      deployment_id: Optional[UUID] = Field(
          default=None,
          title="The ID of the deployment that triggered the pipeline run.",
      )
  
  
  class PipelineRunRequest(ProjectScopedRequest):
      """Request model for pipeline runs."""
  
      name: str = Field(
          title="The name of the pipeline run.",
          max_length=STR_FIELD_MAX_LENGTH,
      )
      snapshot: UUID = Field(
          title="The snapshot associated with the pipeline run."
      )
      orchestrator_run_id: Optional[str] = Field(
          title="The orchestrator run ID.",
          max_length=STR_FIELD_MAX_LENGTH,
          default=None,
      )
      start_time: Optional[datetime] = Field(
          title="The start time of the pipeline run.",
          default=None,
      )
      end_time: Optional[datetime] = Field(
          title="The end time of the pipeline run.",
          default=None,
      )
      status: ExecutionStatus = Field(
          title="The status of the pipeline run.",
      )
      status_reason: Optional[str] = Field(
          title="The reason for the status of the pipeline run.",
          default=None,
          max_length=STR_FIELD_MAX_LENGTH,
      )
      orchestrator_environment: Dict[str, Any] = Field(
          default={},
          title=(
              "Environment of the orchestrator that executed this pipeline run "
              "(OS, Python version, etc.)."
          ),
      )
      trigger_execution_id: Optional[UUID] = Field(
          default=None,
          title="ID of the trigger execution that triggered this run.",
      )
      trigger_info: Optional[PipelineRunTriggerInfo] = Field(
          default=None,
          title="Trigger information for the pipeline run.",
      )
      tags: Optional[List[Union[str, Tag]]] = Field(
          default=None,
          title="Tags of the pipeline run.",
      )
      logs: Optional[LogsRequest] = Field(
          default=None,
          title="Logs of the pipeline run.",
      )
  
      @property
      def is_placeholder_request(self) -> bool:
          """Whether the request is a placeholder request.
  
          Returns:
              Whether the request is a placeholder request.
          """
          return self.status in {
              ExecutionStatus.INITIALIZING,
              ExecutionStatus.PROVISIONING,
          }
  
      model_config = ConfigDict(protected_namespaces=())
  
  
  # ------------------ Update Model ------------------
  
  
  class PipelineRunUpdate(BaseUpdate):
      """Pipeline run update model."""
  
      status: Optional[ExecutionStatus] = None
      status_reason: Optional[str] = Field(
          default=None,
          title="The reason for the status of the pipeline run.",
          max_length=STR_FIELD_MAX_LENGTH,
      )
      end_time: Optional[datetime] = None
      is_finished: Optional[bool] = Field(
          default=None,
          title="Whether the pipeline run is finished.",
      )
      orchestrator_run_id: Optional[str] = None
      # TODO: we should maybe have a different update model here, the upper
      #  three attributes should only be for internal use
      add_tags: Optional[List[str]] = Field(
          default=None, title="New tags to add to the pipeline run."
      )
      remove_tags: Optional[List[str]] = Field(
          default=None, title="Tags to remove from the pipeline run."
      )
      add_logs: Optional[List[LogsRequest]] = Field(
          default=None, title="New logs to add to the pipeline run."
      )
  
      model_config = ConfigDict(protected_namespaces=())
  
  
  # ------------------ Response Model ------------------
  
  
  class PipelineRunResponseBody(ProjectScopedResponseBody):
      """Response body for pipeline runs."""
  
      status: ExecutionStatus = Field(
          title="The status of the pipeline run.",
      )
      in_progress: bool = Field(
          title="Whether the pipeline run is in progress.",
      )
      status_reason: Optional[str] = Field(
          default=None,
          title="The reason for the status of the pipeline run.",
      )
      index: int = Field(
          title="The unique index of the run within the pipeline."
      )
  
      model_config = ConfigDict(protected_namespaces=())
  
  
  class PipelineRunResponseMetadata(ProjectScopedResponseMetadata):
      """Response metadata for pipeline runs."""
  
      __zenml_skip_dehydration__: ClassVar[List[str]] = [
          "run_metadata",
          "config",
          "client_environment",
          "orchestrator_environment",
      ]
  
      run_metadata: Dict[str, MetadataType] = Field(
          default={},
          title="Metadata associated with this pipeline run.",
      )
      config: PipelineConfiguration = Field(
          title="The pipeline configuration used for this pipeline run.",
      )
      start_time: Optional[datetime] = Field(
          title="The start time of the pipeline run.",
          default=None,
      )
      end_time: Optional[datetime] = Field(
          title="The end time of the pipeline run.",
          default=None,
      )
      client_environment: Dict[str, Any] = Field(
          default={},
          title=(
              "Environment of the client that initiated this pipeline run "
              "(OS, Python version, etc.)."
          ),
      )
      orchestrator_environment: Dict[str, Any] = Field(
          default={},
          title=(
              "Environment of the orchestrator that executed this pipeline run "
              "(OS, Python version, etc.)."
          ),
      )
      orchestrator_run_id: Optional[str] = Field(
          title="The orchestrator run ID.",
          max_length=STR_FIELD_MAX_LENGTH,
          default=None,
      )
      code_path: Optional[str] = Field(
          default=None,
          title="Optional path where the code is stored in the artifact store.",
      )
      template_id: Optional[UUID] = Field(
          default=None,
          description="DEPRECATED: Template used for the pipeline run.",
          deprecated=True,
      )
      is_templatable: bool = Field(
          default=False,
          description="Whether a template can be created from this run.",
      )
      trigger_info: Optional[PipelineRunTriggerInfo] = Field(
          default=None,
          title="Trigger information for the pipeline run.",
      )
  
  
  class PipelineRunResponseResources(ProjectScopedResponseResources):
      """Class for all resource models associated with the pipeline run entity."""
  
      snapshot: Optional["PipelineSnapshotResponse"] = None
      source_snapshot: Optional["PipelineSnapshotResponse"] = None
      stack: Optional["StackResponse"] = Field(
          default=None, title="The stack that was used for this run."
      )
      pipeline: Optional["PipelineResponse"] = Field(
          default=None, title="The pipeline this run belongs to."
      )
      build: Optional["PipelineBuildResponse"] = Field(
          default=None, title="The pipeline build that was used for this run."
      )
      schedule: Optional["ScheduleResponse"] = Field(
          default=None, title="The schedule that was used for this run."
      )
      code_reference: Optional["CodeReferenceResponse"] = Field(
          default=None, title="The code reference that was used for this run."
      )
      trigger_execution: Optional["TriggerExecutionResponse"] = Field(
          default=None, title="The trigger execution that triggered this run."
      )
      model_version: Optional[ModelVersionResponse] = None
      tags: List[TagResponse] = Field(
          title="Tags associated with the pipeline run.",
      )
      log_collection: Optional[List["LogsResponse"]] = Field(
          title="Logs associated with this pipeline run.",
          default=None,
      )
      visualizations: List["CuratedVisualizationResponse"] = Field(
          default=[],
          title="Curated visualizations associated with the pipeline run.",
      )
  
      # TODO: In Pydantic v2, the `model_` is a protected namespaces for all
      #  fields defined under base models. If not handled, this raises a warning.
      #  It is possible to suppress this warning message with the following
      #  configuration, however the ultimate solution is to rename these fields.
      #  Even though they do not cause any problems right now, if we are not
      #  careful we might overwrite some fields protected by pydantic.
      model_config = ConfigDict(protected_namespaces=())
  
  
  class PipelineRunResponse(
      ProjectScopedResponse[
          PipelineRunResponseBody,
          PipelineRunResponseMetadata,
          PipelineRunResponseResources,
      ]
  ):
      """Response model for pipeline runs."""
  
      name: str = Field(
          title="The name of the pipeline run.",
          max_length=STR_FIELD_MAX_LENGTH,
      )
  
      def get_hydrated_version(self) -> "PipelineRunResponse":
          """Get the hydrated version of this pipeline run.
  
          Returns:
              an instance of the same entity with the metadata field attached.
          """
          from zenml.client import Client
  
          return Client().zen_store.get_run(self.id)
  
      # Helper methods
      @property
      def artifact_versions(self) -> List["ArtifactVersionResponse"]:
          """Get all artifact versions that are outputs of steps of this run.
  
          Returns:
              All output artifact versions of this run (including cached ones).
          """
          from zenml.artifacts.utils import (
              get_artifacts_versions_of_pipeline_run,
          )
  
          return get_artifacts_versions_of_pipeline_run(self)
  
      @property
      def produced_artifact_versions(self) -> List["ArtifactVersionResponse"]:
          """Get all artifact versions produced during this pipeline run.
  
          Returns:
              A list of all artifact versions produced during this pipeline run.
          """
          from zenml.artifacts.utils import (
              get_artifacts_versions_of_pipeline_run,
          )
  
          return get_artifacts_versions_of_pipeline_run(self, only_produced=True)
  
      # Body and metadata properties
      @property
      def status(self) -> ExecutionStatus:
          """The `status` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().status
  
      @property
      def index(self) -> int:
          """The `index` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().index
  
      @property
      def run_metadata(self) -> Dict[str, MetadataType]:
          """The `run_metadata` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().run_metadata
  
      @property
      def steps(self) -> Dict[str, "StepRunResponse"]:
          """The `steps` property.
  
          Returns:
              the value of the property.
          """
          from zenml.client import Client
  
          return {
              step.name: step
              for step in pagination_utils.depaginate(
                  Client().list_run_steps,
                  pipeline_run_id=self.id,
                  project=self.project_id,
                  exclude_retried=True,
              )
          }
  
      @property
      def config(self) -> PipelineConfiguration:
          """The `config` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().config
  
      @property
      def start_time(self) -> Optional[datetime]:
          """The `start_time` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().start_time
  
      @property
      def end_time(self) -> Optional[datetime]:
          """The `end_time` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().end_time
  
      @property
      def in_progress(self) -> bool:
          """The `in_progress` property.
  
          Returns:
              the value of the property.
          """
          return self.get_body().in_progress
  
      @property
      def client_environment(self) -> Dict[str, Any]:
          """The `client_environment` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().client_environment
  
      @property
      def orchestrator_environment(self) -> Dict[str, Any]:
          """The `orchestrator_environment` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().orchestrator_environment
  
      @property
      def orchestrator_run_id(self) -> Optional[str]:
          """The `orchestrator_run_id` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().orchestrator_run_id
  
      @property
      def code_path(self) -> Optional[str]:
          """The `code_path` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().code_path
  
      @property
      def template_id(self) -> Optional[UUID]:
          """The `template_id` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().template_id
  
      @property
      def is_templatable(self) -> bool:
          """The `is_templatable` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().is_templatable
  
      @property
      def trigger_info(self) -> Optional[PipelineRunTriggerInfo]:
          """The `trigger_info` property.
  
          Returns:
              the value of the property.
          """
          return self.get_metadata().trigger_info
  
      @property
      def triggered_by_deployment(self) -> bool:
          """The `triggered_by_deployment` property.
  
          Returns:
              the value of the property.
          """
          return (
              self.trigger_info is not None
              and self.trigger_info.deployment_id is not None
          )
  
      @property
      def snapshot(self) -> Optional["PipelineSnapshotResponse"]:
          """The `snapshot` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().snapshot
  
      @property
      def source_snapshot(self) -> Optional["PipelineSnapshotResponse"]:
          """The `source_snapshot` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().source_snapshot
  
      @property
      def stack(self) -> Optional["StackResponse"]:
          """The `stack` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().stack
  
      @property
      def pipeline(self) -> Optional["PipelineResponse"]:
          """The `pipeline` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().pipeline
  
      @property
      def build(self) -> Optional["PipelineBuildResponse"]:
          """The `build` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().build
  
      @property
      def schedule(self) -> Optional["ScheduleResponse"]:
          """The `schedule` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().schedule
  
      @property
      def trigger_execution(self) -> Optional["TriggerExecutionResponse"]:
          """The `trigger_execution` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().trigger_execution
  
      @property
      def code_reference(self) -> Optional["CodeReferenceResponse"]:
          """The `schedule` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().code_reference
  
      @property
      def model_version(self) -> Optional[ModelVersionResponse]:
          """The `model_version` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().model_version
  
      @property
      def tags(self) -> List[TagResponse]:
          """The `tags` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().tags
  
      @property
      def log_collection(self) -> Optional[List["LogsResponse"]]:
          """The `log_collection` property.
  
          Returns:
              the value of the property.
          """
          return self.get_resources().log_collection
  
  
  # ------------------ Filter Model ------------------
  
  
  class PipelineRunFilter(
      ProjectScopedFilter, TaggableFilter, RunMetadataFilterMixin
  ):
      """Model to enable advanced filtering of all pipeline runs."""
  
      CUSTOM_SORTING_OPTIONS: ClassVar[List[str]] = [
          *ProjectScopedFilter.CUSTOM_SORTING_OPTIONS,
          *TaggableFilter.CUSTOM_SORTING_OPTIONS,
          *RunMetadataFilterMixin.CUSTOM_SORTING_OPTIONS,
          "stack",
          "pipeline",
          "model",
          "model_version",
      ]
      FILTER_EXCLUDE_FIELDS: ClassVar[List[str]] = [
          *ProjectScopedFilter.FILTER_EXCLUDE_FIELDS,
          *TaggableFilter.FILTER_EXCLUDE_FIELDS,
          *RunMetadataFilterMixin.FILTER_EXCLUDE_FIELDS,
          "unlisted",
          "code_repository_id",
          "build_id",
          "schedule_id",
          "stack_id",
          "template_id",
          "source_snapshot_id",
          "pipeline",
          "stack",
          "code_repository",
          "model",
          "stack_component",
          "pipeline_name",
          "templatable",
          "triggered_by_step_run_id",
          "triggered_by_deployment_id",
          "linked_to_model_version_id",
      ]
      CLI_EXCLUDE_FIELDS = [
          *ProjectScopedFilter.CLI_EXCLUDE_FIELDS,
          *TaggableFilter.CLI_EXCLUDE_FIELDS,
          *RunMetadataFilterMixin.CLI_EXCLUDE_FIELDS,
      ]
      API_MULTI_INPUT_PARAMS: ClassVar[List[str]] = [
          *ProjectScopedFilter.API_MULTI_INPUT_PARAMS,
          *TaggableFilter.API_MULTI_INPUT_PARAMS,
          *RunMetadataFilterMixin.API_MULTI_INPUT_PARAMS,
      ]
  
      name: Optional[str] = Field(
          default=None,
          description="Name of the Pipeline Run",
      )
      index: Optional[int] = Field(
          default=None,
          description="The unique index of the run within the pipeline.",
      )
      orchestrator_run_id: Optional[str] = Field(
          default=None,
          description="Name of the Pipeline Run within the orchestrator",
      )
      pipeline_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Pipeline associated with the Pipeline Run",
          union_mode="left_to_right",
      )
      stack_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Stack used for the Pipeline Run",
          union_mode="left_to_right",
      )
      schedule_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Schedule that triggered the Pipeline Run",
          union_mode="left_to_right",
      )
      build_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Build used for the Pipeline Run",
          union_mode="left_to_right",
      )
      snapshot_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Snapshot used for the Pipeline Run",
          union_mode="left_to_right",
      )
      code_repository_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Code repository used for the Pipeline Run",
          union_mode="left_to_right",
      )
      template_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="DEPRECATED: Template used for the pipeline run.",
          union_mode="left_to_right",
          deprecated=True,
      )
      source_snapshot_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Source snapshot used for the pipeline run.",
          union_mode="left_to_right",
      )
      model_version_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Model version associated with the pipeline run.",
          union_mode="left_to_right",
      )
      linked_to_model_version_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Filter by model version linked to the pipeline run. "
          "The difference to `model_version_id` is that this filter will "
          "not only include pipeline runs which are directly linked to the model "
          "version, but also if any step run is linked to the model version.",
          union_mode="left_to_right",
      )
      status: Optional[str] = Field(
          default=None,
          description="Name of the Pipeline Run",
      )
      in_progress: Optional[bool] = Field(
          default=None,
          description="Whether the pipeline run is in progress.",
      )
      start_time: Optional[Union[datetime, str]] = Field(
          default=None,
          description="Start time for this run",
          union_mode="left_to_right",
      )
      end_time: Optional[Union[datetime, str]] = Field(
          default=None,
          description="End time for this run",
          union_mode="left_to_right",
      )
      unlisted: Optional[bool] = None
      # TODO: Remove once frontend is ready for it. This is replaced by the more
      #   generic `pipeline` filter below.
      pipeline_name: Optional[str] = Field(
          default=None,
          description="Name of the pipeline associated with the run",
      )
      pipeline: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Name/ID of the pipeline associated with the run.",
      )
      stack: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Name/ID of the stack associated with the run.",
      )
      code_repository: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Name/ID of the code repository associated with the run.",
      )
      model: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Name/ID of the model associated with the run.",
      )
      stack_component: Optional[Union[UUID, str]] = Field(
          default=None,
          description="Name/ID of the stack component associated with the run.",
      )
      templatable: Optional[bool] = Field(
          default=None, description="Whether the run is templatable."
      )
      triggered_by_step_run_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="The ID of the step run that triggered this pipeline run.",
          union_mode="left_to_right",
      )
      triggered_by_deployment_id: Optional[Union[UUID, str]] = Field(
          default=None,
          description="The ID of the deployment that triggered this pipeline run.",
          union_mode="left_to_right",
      )
      model_config = ConfigDict(protected_namespaces=())
  
      def get_custom_filters(
          self,
          table: Type["AnySchema"],
      ) -> List["ColumnElement[bool]"]:
          """Get custom filters.
  
          Args:
              table: The query table.
  
          Returns:
              A list of custom filters.
          """
          custom_filters = super().get_custom_filters(table)
  
          from sqlmodel import and_, col, or_
  
          from zenml.zen_stores.schemas import (
              CodeReferenceSchema,
              CodeRepositorySchema,
              DeploymentSchema,
              ModelSchema,
              ModelVersionPipelineRunSchema,
              ModelVersionSchema,
              PipelineBuildSchema,
              PipelineRunSchema,
              PipelineSchema,
              PipelineSnapshotSchema,
              ScheduleSchema,
              StackComponentSchema,
              StackCompositionSchema,
              StackSchema,
              StepRunSchema,
          )
  
          if self.unlisted is not None:
              if self.unlisted is True:
                  unlisted_filter = PipelineRunSchema.pipeline_id.is_(None)  # type: ignore[union-attr]
              else:
                  unlisted_filter = PipelineRunSchema.pipeline_id.is_not(None)  # type: ignore[union-attr]
              custom_filters.append(unlisted_filter)
  
          if self.code_repository_id:
              code_repo_filter = and_(
                  PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                  PipelineSnapshotSchema.code_reference_id
                  == CodeReferenceSchema.id,
                  CodeReferenceSchema.code_repository_id
                  == self.code_repository_id,
              )
              custom_filters.append(code_repo_filter)
  
          if self.stack_id:
              stack_filter = and_(
                  PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                  PipelineSnapshotSchema.stack_id == StackSchema.id,
                  StackSchema.id == self.stack_id,
              )
              custom_filters.append(stack_filter)
  
          if self.schedule_id:
              schedule_filter = and_(
                  PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                  PipelineSnapshotSchema.schedule_id == ScheduleSchema.id,
                  ScheduleSchema.id == self.schedule_id,
              )
              custom_filters.append(schedule_filter)
  
          if self.build_id:
              pipeline_build_filter = and_(
                  PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                  PipelineSnapshotSchema.build_id == PipelineBuildSchema.id,
                  PipelineBuildSchema.id == self.build_id,
              )
              custom_filters.append(pipeline_build_filter)
  
          if self.template_id:
              run_template_filter = and_(
                  PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                  PipelineSnapshotSchema.template_id == self.template_id,
              )
              custom_filters.append(run_template_filter)
  
          if self.source_snapshot_id:
              source_snapshot_filter = and_(
                  PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                  PipelineSnapshotSchema.source_snapshot_id
                  == self.source_snapshot_id,
              )
              custom_filters.append(source_snapshot_filter)
  
          if self.pipeline:
              pipeline_filter = and_(
                  PipelineRunSchema.pipeline_id == PipelineSchema.id,
                  self.generate_name_or_id_query_conditions(
                      value=self.pipeline, table=PipelineSchema
                  ),
              )
              custom_filters.append(pipeline_filter)
  
          if self.stack:
              stack_filter = and_(
                  PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                  PipelineSnapshotSchema.stack_id == StackSchema.id,
                  self.generate_name_or_id_query_conditions(
                      value=self.stack,
                      table=StackSchema,
                  ),
              )
              custom_filters.append(stack_filter)
  
          if self.code_repository:
              code_repo_filter = and_(
                  PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                  PipelineSnapshotSchema.code_reference_id
                  == CodeReferenceSchema.id,
                  CodeReferenceSchema.code_repository_id
                  == CodeRepositorySchema.id,
                  self.generate_name_or_id_query_conditions(
                      value=self.code_repository,
                      table=CodeRepositorySchema,
                  ),
              )
              custom_filters.append(code_repo_filter)
  
          if self.model:
              model_filter = and_(
                  PipelineRunSchema.model_version_id == ModelVersionSchema.id,
                  ModelVersionSchema.model_id == ModelSchema.id,
                  self.generate_name_or_id_query_conditions(
                      value=self.model, table=ModelSchema
                  ),
              )
              custom_filters.append(model_filter)
  
          if self.stack_component:
              component_filter = and_(
                  PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                  PipelineSnapshotSchema.stack_id == StackSchema.id,
                  StackSchema.id == StackCompositionSchema.stack_id,
                  StackCompositionSchema.component_id == StackComponentSchema.id,
                  self.generate_name_or_id_query_conditions(
                      value=self.stack_component,
                      table=StackComponentSchema,
                  ),
              )
              custom_filters.append(component_filter)
  
          if self.pipeline_name:
              pipeline_name_filter = and_(
                  PipelineRunSchema.pipeline_id == PipelineSchema.id,
                  self.generate_custom_query_conditions_for_column(
                      value=self.pipeline_name,
                      table=PipelineSchema,
                      column="name",
                  ),
              )
              custom_filters.append(pipeline_name_filter)
  
          if self.templatable is not None:
              if self.templatable is True:
                  templatable_filter = and_(
                      # The following condition is not perfect as it does not
                      # consider stacks with custom flavor components or local
                      # components, but the best we can do currently with our
                      # table columns.
                      PipelineRunSchema.snapshot_id == PipelineSnapshotSchema.id,
                      PipelineSnapshotSchema.build_id == PipelineBuildSchema.id,
                      col(PipelineBuildSchema.is_local).is_(False),
                      col(PipelineBuildSchema.stack_id).is_not(None),
                  )
              else:
                  templatable_filter = or_(
                      col(PipelineRunSchema.snapshot_id).is_(None),
                      and_(
                          PipelineRunSchema.snapshot_id
                          == PipelineSnapshotSchema.id,
                          col(PipelineSnapshotSchema.build_id).is_(None),
                      ),
                      and_(
                          PipelineRunSchema.snapshot_id
                          == PipelineSnapshotSchema.id,
                          PipelineSnapshotSchema.build_id
                          == PipelineBuildSchema.id,
                          or_(
                              col(PipelineBuildSchema.is_local).is_(True),
  