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.
  """Definition of a ZenML pipeline."""
  
  import copy
  import hashlib
  import inspect
  from contextlib import contextmanager, nullcontext
  from pathlib import Path
  from typing import (
      TYPE_CHECKING,
      Any,
      Callable,
      Dict,
      Iterator,
      List,
      Mapping,
      Optional,
      Sequence,
      Set,
      Tuple,
      Type,
      TypeVar,
      Union,
  )
  from uuid import UUID
  
  import yaml
  from pydantic import BaseModel, ConfigDict, ValidationError, create_model
  from typing_extensions import Self
  
  from zenml.analytics.enums import AnalyticsEvent
  from zenml.analytics.utils import track_handler
  from zenml.client import Client
  from zenml.config.compiler import Compiler
  from zenml.config.pipeline_configurations import (
      PipelineConfiguration,
      PipelineConfigurationUpdate,
  )
  from zenml.config.pipeline_run_configuration import PipelineRunConfiguration
  from zenml.config.pipeline_spec import PipelineSpec
  from zenml.config.schedule import Schedule
  from zenml.config.step_configurations import StepConfigurationUpdate
  from zenml.enums import StackComponentType
  from zenml.exceptions import EntityExistsError
  from zenml.execution.pipeline.utils import (
      should_prevent_pipeline_execution,
      submit_pipeline,
  )
  from zenml.hooks.hook_validators import resolve_and_validate_hook
  from zenml.logger import get_logger
  from zenml.models import (
      CodeReferenceRequest,
      DeploymentResponse,
      PipelineBuildBase,
      PipelineBuildResponse,
      PipelineRequest,
      PipelineResponse,
      PipelineRunResponse,
      PipelineSnapshotBase,
      PipelineSnapshotRequest,
      PipelineSnapshotResponse,
      RunTemplateResponse,
      ScheduleRequest,
  )
  from zenml.pipelines import build_utils
  from zenml.pipelines.compilation_context import PipelineCompilationContext
  from zenml.pipelines.run_utils import (
      create_placeholder_run,
      upload_notebook_cell_code_if_necessary,
  )
  from zenml.stack import Stack
  from zenml.steps import BaseStep
  from zenml.steps.entrypoint_function_utils import StepArtifact
  from zenml.steps.step_invocation import StepInvocation
  from zenml.steps.utils import get_unique_step_output_names
  from zenml.utils import (
      code_repository_utils,
      code_utils,
      dashboard_utils,
      dict_utils,
      env_utils,
      pydantic_utils,
      settings_utils,
      source_utils,
      yaml_utils,
  )
  from zenml.utils.logging_utils import (
      is_pipeline_logging_enabled,
      setup_run_logging,
  )
  from zenml.utils.string_utils import format_name_template
  from zenml.utils.tag_utils import Tag
  
  if TYPE_CHECKING:
      from zenml.artifacts.external_artifact import ExternalArtifact
      from zenml.client_lazy_loader import ClientLazyLoader
      from zenml.config.base_settings import SettingsOrDict
      from zenml.config.cache_policy import CachePolicyOrString
      from zenml.config.retry_config import StepRetryConfig
      from zenml.config.source import Source
      from zenml.enums import ExecutionMode
      from zenml.model.lazy_load import ModelVersionDataLazyLoader
      from zenml.model.model import Model
      from zenml.models import ArtifactVersionResponse
      from zenml.types import HookSpecification, InitHookSpecification
  
      StepConfigurationUpdateOrDict = Union[
          Dict[str, Any], StepConfigurationUpdate
      ]
  
  logger = get_logger(__name__)
  
  F = TypeVar("F", bound=Callable[..., Any])
  
  
  class Pipeline:
      """ZenML pipeline class."""
  
      def __init__(
          self,
          *,
          name: str,
          entrypoint: F,
          enable_cache: Optional[bool] = None,
          enable_artifact_metadata: Optional[bool] = None,
          enable_artifact_visualization: Optional[bool] = None,
          enable_step_logs: Optional[bool] = None,
          environment: Optional[Dict[str, Any]] = None,
          secrets: Optional[List[Union[UUID, str]]] = None,
          enable_pipeline_logs: Optional[bool] = None,
          settings: Optional[Mapping[str, "SettingsOrDict"]] = None,
          tags: Optional[List[Union[str, "Tag"]]] = None,
          extra: Optional[Dict[str, Any]] = None,
          on_failure: Optional["HookSpecification"] = None,
          on_success: Optional["HookSpecification"] = None,
          on_init: Optional["InitHookSpecification"] = None,
          on_init_kwargs: Optional[Dict[str, Any]] = None,
          on_cleanup: Optional["HookSpecification"] = None,
          model: Optional["Model"] = None,
          retry: Optional["StepRetryConfig"] = None,
          substitutions: Optional[Dict[str, str]] = None,
          execution_mode: Optional["ExecutionMode"] = None,
          cache_policy: Optional["CachePolicyOrString"] = None,
          **kwargs: Any,
      ) -> None:
          """Initializes a pipeline.
  
          Args:
              name: The name of the pipeline.
              entrypoint: The entrypoint function of the pipeline.
              enable_cache: If caching should be enabled for this pipeline.
              enable_artifact_metadata: If artifact metadata should be enabled for
                  this pipeline.
              enable_artifact_visualization: If artifact visualization should be
                  enabled for this pipeline.
              enable_step_logs: If step logs should be enabled for this pipeline.
              environment: Environment variables to set when running this
                  pipeline.
              secrets: Secrets to set as environment variables when running this
                  pipeline.
              enable_pipeline_logs: If pipeline logs should be enabled for this pipeline.
              settings: Settings for this pipeline.
              tags: Tags to apply to runs of this pipeline.
              extra: Extra configurations for this pipeline.
              on_failure: Callback function in event of failure of the step. Can
                  be a function with a single argument of type `BaseException`, or
                  a source path to such a function (e.g. `module.my_function`).
              on_success: Callback function in event of success of the step. Can
                  be a function with no arguments, or a source path to such a
                  function (e.g. `module.my_function`).
              on_init: Callback function to run on initialization of the pipeline.
                  Can be a function with no arguments, or a source path to such a
                  function (e.g. `module.my_function`) if the function returns a
                  value, it will be stored as the pipeline state.
              on_init_kwargs: Arguments for the init hook.
              on_cleanup: Callback function to run on cleanup of the pipeline. Can
                  be a function with no arguments, or a source path to such a
                  function with no arguments (e.g. `module.my_function`).
              model: configuration of the model in the Model Control Plane.
              retry: Retry configuration for the pipeline steps.
              substitutions: Extra substitutions for pipeline run, model and
                  artifact name placeholders.
              execution_mode: The execution mode of the pipeline.
              cache_policy: Cache policy for this pipeline.
              **kwargs: Additional keyword arguments.
          """
          self._invocations: Dict[str, StepInvocation] = {}
          self._run_args: Dict[str, Any] = {}
  
          self._configuration = PipelineConfiguration(
              name=name,
          )
          self._from_config_file: Dict[str, Any] = {}
          with self.__suppress_configure_warnings__():
              self.configure(
                  enable_cache=enable_cache,
                  enable_artifact_metadata=enable_artifact_metadata,
                  enable_artifact_visualization=enable_artifact_visualization,
                  enable_step_logs=enable_step_logs,
                  environment=environment,
                  secrets=secrets,
                  enable_pipeline_logs=enable_pipeline_logs,
                  settings=settings,
                  tags=tags,
                  extra=extra,
                  on_failure=on_failure,
                  on_success=on_success,
                  on_init=on_init,
                  on_init_kwargs=on_init_kwargs,
                  on_cleanup=on_cleanup,
                  model=model,
                  retry=retry,
                  substitutions=substitutions,
                  execution_mode=execution_mode,
                  cache_policy=cache_policy,
              )
          self.entrypoint = entrypoint
          self._parameters: Dict[str, Any] = {}
          self._output_artifacts: List[StepArtifact] = []
  
          self.__suppress_warnings_flag__ = False
  
      @property
      def name(self) -> str:
          """The name of the pipeline.
  
          Returns:
              The name of the pipeline.
          """
          return self.configuration.name
  
      @property
      def is_dynamic(self) -> bool:
          """If the pipeline is dynamic.
  
          Returns:
              If the pipeline is dynamic.
          """
          return False
  
      @property
      def enable_cache(self) -> Optional[bool]:
          """If caching is enabled for the pipeline.
  
          Returns:
              If caching is enabled for the pipeline.
          """
          return self.configuration.enable_cache
  
      @property
      def configuration(self) -> PipelineConfiguration:
          """The configuration of the pipeline.
  
          Returns:
              The configuration of the pipeline.
          """
          return self._configuration
  
      @property
      def invocations(self) -> Dict[str, StepInvocation]:
          """Returns the step invocations of this pipeline.
  
          This dictionary will only be populated once the pipeline has been
          called.
  
          Returns:
              The step invocations.
          """
          return self._invocations
  
      def resolve(self) -> "Source":
          """Resolves the pipeline.
  
          Returns:
              The pipeline source.
          """
          return source_utils.resolve(self.entrypoint, skip_validation=True)
  
      @property
      def source_object(self) -> Any:
          """The source object of this pipeline.
  
          Returns:
              The source object of this pipeline.
          """
          return self.entrypoint
  
      @property
      def source_code(self) -> str:
          """The source code of this pipeline.
  
          Returns:
              The source code of this pipeline.
          """
          return inspect.getsource(self.source_object)
  
      @property
      def model(self) -> "PipelineResponse":
          """Gets the registered pipeline model for this instance.
  
          Returns:
              The registered pipeline model.
  
          Raises:
              RuntimeError: If the pipeline has not been registered yet.
          """
          pipelines = Client().list_pipelines(name=self.name)
          if len(pipelines) == 1:
              return pipelines.items[0]
  
          raise RuntimeError(
              f"Cannot get the model of pipeline '{self.name}' because it has "
              f"not been registered yet. Please ensure that the pipeline has "
              f"been run or built and try again."
          )
  
      @contextmanager
      def __suppress_configure_warnings__(self) -> Iterator[Any]:
          """Context manager to suppress warnings in `Pipeline.configure(...)`.
  
          Used to suppress warnings when called from inner code and not user-facing code.
  
          Yields:
              Nothing.
          """
          self.__suppress_warnings_flag__ = True
          yield
          self.__suppress_warnings_flag__ = False
  
      def configure(
          self,
          enable_cache: Optional[bool] = None,
          enable_artifact_metadata: Optional[bool] = None,
          enable_artifact_visualization: Optional[bool] = None,
          enable_step_logs: Optional[bool] = None,
          environment: Optional[Dict[str, Any]] = None,
          secrets: Optional[Sequence[Union[UUID, str]]] = None,
          enable_pipeline_logs: Optional[bool] = None,
          settings: Optional[Mapping[str, "SettingsOrDict"]] = None,
          tags: Optional[List[Union[str, "Tag"]]] = None,
          extra: Optional[Dict[str, Any]] = None,
          on_failure: Optional["HookSpecification"] = None,
          on_success: Optional["HookSpecification"] = None,
          on_init: Optional["InitHookSpecification"] = None,
          on_init_kwargs: Optional[Dict[str, Any]] = None,
          on_cleanup: Optional["HookSpecification"] = None,
          model: Optional["Model"] = None,
          retry: Optional["StepRetryConfig"] = None,
          parameters: Optional[Dict[str, Any]] = None,
          substitutions: Optional[Dict[str, str]] = None,
          execution_mode: Optional["ExecutionMode"] = None,
          cache_policy: Optional["CachePolicyOrString"] = None,
          merge: bool = True,
      ) -> Self:
          """Configures the pipeline.
  
          Configuration merging example:
          * `merge==True`:
              pipeline.configure(extra={"key1": 1})
              pipeline.configure(extra={"key2": 2}, merge=True)
              pipeline.configuration.extra # {"key1": 1, "key2": 2}
          * `merge==False`:
              pipeline.configure(extra={"key1": 1})
              pipeline.configure(extra={"key2": 2}, merge=False)
              pipeline.configuration.extra # {"key2": 2}
  
          Args:
              enable_cache: If caching should be enabled for this pipeline.
              enable_artifact_metadata: If artifact metadata should be enabled for
                  this pipeline.
              enable_artifact_visualization: If artifact visualization should be
                  enabled for this pipeline.
              enable_step_logs: If step logs should be enabled for this pipeline.
              environment: Environment variables to set when running this
                  pipeline.
              secrets: Secrets to set as environment variables when running this
                  pipeline.
              settings: Settings for this pipeline.
              enable_pipeline_logs: If pipeline logs should be enabled for this pipeline.
              settings: settings for this pipeline.
              tags: Tags to apply to runs of this pipeline.
              extra: Extra configurations for this pipeline.
              on_failure: Callback function in event of failure of the step. Can
                  be a function with a single argument of type `BaseException`, or
                  a source path to such a function (e.g. `module.my_function`).
              on_success: Callback function in event of success of the step. Can
                  be a function with no arguments, or a source path to such a
                  function (e.g. `module.my_function`).
              on_init: Callback function to run on initialization of the pipeline.
                  Can be a function with no arguments, or a source path to such a
                  function (e.g. `module.my_function`) if the function returns a
                  value, it will be stored as the pipeline state.
              on_init_kwargs: Arguments for the init hook.
              on_cleanup: Callback function to run on cleanup of the pipeline. Can
                  be a function with no arguments, or a source path to such a
                  function with no arguments (e.g. `module.my_function`).
              model: configuration of the model version in the Model Control Plane.
              retry: Retry configuration for the pipeline steps.
              parameters: input parameters for the pipeline.
              substitutions: Extra substitutions for pipeline run, model and
                  artifact name placeholders.
              execution_mode: The execution mode of the pipeline.
              cache_policy: Cache policy for this pipeline.
              merge: If `True`, will merge the given dictionary configurations
                  like `extra` and `settings` with existing
                  configurations. If `False` the given configurations will
                  overwrite all existing ones. See the general description of this
                  method for an example.
  
          Returns:
              The pipeline instance that this method was called on.
  
          Raises:
              ValueError: If on_init_kwargs is provided but on_init is not and
                  the init hook source is found in the current pipeline
                  configuration.
          """
          failure_hook_source = None
          if on_failure:
              # string of on_failure hook function to be used for this pipeline
              failure_hook_source, _ = resolve_and_validate_hook(
                  on_failure, allow_exception_arg=True
              )
  
          success_hook_source = None
          if on_success:
              # string of on_success hook function to be used for this pipeline
              success_hook_source, _ = resolve_and_validate_hook(on_success)
  
          init_hook_kwargs = None
          init_hook_source = None
          if on_init or on_init_kwargs:
              if not on_init and self.configuration.init_hook_source:
                  # load the init hook source from the existing configuration if
                  # not provided; this is needed for partial updates
                  on_init = source_utils.load(
                      self.configuration.init_hook_source
                  )
              if not on_init:
                  raise ValueError(
                      "on_init is not provided and no init hook source is found "
                      "in the existing configuration"
                  )
  
              # string of on_init hook function and JSON-able arguments to be used
              # for this pipeline
              init_hook_source, init_hook_kwargs = resolve_and_validate_hook(
                  on_init, on_init_kwargs
              )
  
          cleanup_hook_source = None
          if on_cleanup:
              # string of on_cleanup hook function to be used for this pipeline
              cleanup_hook_source, _ = resolve_and_validate_hook(on_cleanup)
  
          if merge and tags and self._configuration.tags:
              # Merge tags explicitly here as the recursive update later only
              # merges dicts
              tags = self._configuration.tags + tags
  
          if merge and secrets and self._configuration.secrets:
              secrets = self._configuration.secrets + list(secrets)
  
          values = dict_utils.remove_none_values(
              {
                  "enable_cache": enable_cache,
                  "enable_artifact_metadata": enable_artifact_metadata,
                  "enable_artifact_visualization": enable_artifact_visualization,
                  "enable_step_logs": enable_step_logs,
                  "environment": environment,
                  "secrets": secrets,
                  "enable_pipeline_logs": enable_pipeline_logs,
                  "settings": settings,
                  "tags": tags,
                  "extra": extra,
                  "failure_hook_source": failure_hook_source,
                  "success_hook_source": success_hook_source,
                  "init_hook_source": init_hook_source,
                  "init_hook_kwargs": init_hook_kwargs,
                  "cleanup_hook_source": cleanup_hook_source,
                  "model": model,
                  "retry": retry,
                  "parameters": parameters,
                  "substitutions": substitutions,
                  "execution_mode": execution_mode,
                  "cache_policy": cache_policy,
              }
          )
          if not self.__suppress_warnings_flag__:
              to_be_reapplied = []
              for param_, value_ in values.items():
                  if (
                      param_ in PipelineRunConfiguration.model_fields
                      and param_ in self._from_config_file
                      and value_ != self._from_config_file[param_]
                  ):
                      to_be_reapplied.append(
                          (param_, self._from_config_file[param_], value_)
                      )
              if to_be_reapplied:
                  msg = ""
                  reapply_during_run_warning = (
                      "The value of parameter '{name}' has changed from "
                      "'{file_value}' to '{new_value}' set in your configuration "
                      "file.\n"
                  )
                  for name, file_value, new_value in to_be_reapplied:
                      msg += reapply_during_run_warning.format(
                          name=name, file_value=file_value, new_value=new_value
                      )
                  msg += (
                      "Configuration file value will be used during pipeline "
                      "run, so you change will not be efficient. Consider "
                      "updating your configuration file instead."
                  )
                  logger.warning(msg)
  
          config = PipelineConfigurationUpdate(**values)
          self._apply_configuration(config, merge=merge)
          return self
  
      @property
      def required_parameters(self) -> List[str]:
          """List of required parameters for the pipeline entrypoint.
  
          Returns:
              List of required parameters for the pipeline entrypoint.
          """
          signature = inspect.signature(self.entrypoint, follow_wrapped=True)
          return [
              parameter.name
              for parameter in signature.parameters.values()
              if parameter.default is inspect.Parameter.empty
          ]
  
      @property
      def missing_parameters(self) -> List[str]:
          """List of missing parameters for the pipeline entrypoint.
  
          Returns:
              List of missing parameters for the pipeline entrypoint.
          """
          available_parameters = set(self.configuration.parameters or {})
          if params_from_file := self._from_config_file.get("parameters", None):
              available_parameters.update(params_from_file)
  
          return list(set(self.required_parameters) - available_parameters)
  
      @property
      def is_prepared(self) -> bool:
          """If the pipeline is prepared.
  
          Prepared means that the pipeline entrypoint has been called and the
          pipeline is fully defined.
  
          Returns:
              If the pipeline is prepared.
          """
          return len(self.invocations) > 0
  
      def prepare(self, *args: Any, **kwargs: Any) -> None:
          """Prepares the pipeline.
  
          Args:
              *args: Pipeline entrypoint input arguments.
              **kwargs: Pipeline entrypoint input keyword arguments.
          """
          self._clear_state()
  
          kwargs = self._apply_config_parameters(kwargs)
          self._parameters = self._validate_entrypoint_args(*args, **kwargs)
  
          with PipelineCompilationContext(pipeline=self):
              self._prepare_invocations(**self._parameters)
  
      def _validate_entrypoint_args(
          self, *args: Any, **kwargs: Any
      ) -> Dict[str, Any]:
          """Validates the arguments for the pipeline entrypoint.
  
          Args:
              *args: Entrypoint function arguments.
              **kwargs: Entrypoint function keyword arguments.
  
          Raises:
              ValueError: If the arguments are invalid or missing.
  
          Returns:
              The validated arguments.
          """
          try:
              validated_args = pydantic_utils.validate_function_args(
                  self.entrypoint,
                  ConfigDict(arbitrary_types_allowed=False),
                  *args,
                  **kwargs,
              )
          except ValidationError as e:
              raise ValueError(
                  "Invalid or missing pipeline function entrypoint arguments. "
                  "Only JSON serializable inputs are allowed as pipeline inputs. "
                  "Check out the pydantic error above for more details."
              ) from e
  
          return validated_args
  
      def _apply_config_parameters(
          self, kwargs: Dict[str, Any]
      ) -> Dict[str, Any]:
          """Applies the configuration parameters to the code arguments.
  
          Args:
              kwargs: The code arguments to apply the configuration parameters to.
  
          Raises:
              RuntimeError: If different values for the same key are passed in
                  code and configuration.
  
          Returns:
              The merged arguments.
          """
          kwargs = kwargs.copy()
          conflicting_parameters = {}
          config_parameters = self.configuration.parameters or {}
          if from_file_ := self._from_config_file.get("parameters", None):
              config_parameters = dict_utils.recursive_update(
                  config_parameters, from_file_
              )
  
          if config_parameters:
              for k, v_runtime in kwargs.items():
                  if k in config_parameters:
                      v_config = config_parameters[k]
                      if v_config != v_runtime:
                          conflicting_parameters[k] = (v_config, v_runtime)
  
              if conflicting_parameters:
                  is_plural = "s" if len(conflicting_parameters) > 1 else ""
                  msg = f"Configured parameter{is_plural} for the pipeline `{self.name}` conflict{'' if not is_plural else 's'} with parameter{is_plural} passed in runtime:\n"
                  for key, values in conflicting_parameters.items():
                      msg += f"`{key}`: config=`{values[0]}` | runtime=`{values[1]}`\n"
                  msg += """This happens, if you define values for pipeline parameters in configuration file and pass same parameters from the code. Example:
  ```
  # config.yaml
      parameters:
          param_name: value1
              
              
  # pipeline.py
  @pipeline
  def pipeline_(param_name: str):
      step_name()
  
  if __name__=="__main__":
      pipeline_.with_options(config_path="config.yaml")(param_name="value2")
  ```
  To avoid this consider setting pipeline parameters only in one place (config or code).
  """
                  raise RuntimeError(msg)
  
              for k, v_config in config_parameters.items():
                  if k not in kwargs:
                      kwargs[k] = v_config
  
          return kwargs
  
      def _prepare_invocations(self, **kwargs: Any) -> None:
          """Prepares the invocations of the pipeline.
  
          Args:
              **kwargs: Keyword arguments.
          """
          outputs = self._call_entrypoint(**kwargs)
  
          output_artifacts = []
          if isinstance(outputs, StepArtifact):
              output_artifacts = [outputs]
          elif isinstance(outputs, tuple):
              for v in outputs:
                  if isinstance(v, StepArtifact):
                      output_artifacts.append(v)
                  else:
                      logger.debug(
                          "Ignore pipeline output that is not a step artifact: %s",
                          v,
                      )
  
          self._output_artifacts = output_artifacts
  
      def register(self) -> "PipelineResponse":
          """Register the pipeline in the server.
  
          Returns:
              The registered pipeline model.
          """
          client = Client()
  
          def _get() -> PipelineResponse:
              matching_pipelines = client.list_pipelines(
                  name=self.name,
                  size=1,
                  sort_by="desc:created",
              )
  
              if matching_pipelines.total:
                  registered_pipeline = matching_pipelines.items[0]
                  return registered_pipeline
              raise RuntimeError("No matching pipelines found.")
  
          try:
              return _get()
          except RuntimeError:
              request = PipelineRequest(
                  project=client.active_project.id,
                  name=self.name,
              )
  
              try:
                  registered_pipeline = client.zen_store.create_pipeline(
                      pipeline=request
                  )
                  logger.info(
                      "Registered new pipeline: `%s`.",
                      registered_pipeline.name,
                  )
                  return registered_pipeline
              except EntityExistsError:
                  return _get()
  
      def build(
          self,
          settings: Optional[Mapping[str, "SettingsOrDict"]] = None,
          step_configurations: Optional[
              Mapping[str, "StepConfigurationUpdateOrDict"]
          ] = None,
          config_path: Optional[str] = None,
      ) -> Optional["PipelineBuildResponse"]:
          """Builds Docker images for the pipeline.
  
          Args:
              settings: Settings for the pipeline.
              step_configurations: Configurations for steps of the pipeline.
              config_path: Path to a yaml configuration file. This file will
                  be parsed as a
                  `zenml.config.pipeline_configurations.PipelineRunConfiguration`
                  object. Options provided in this file will be overwritten by
                  options provided in code using the other arguments of this
                  method.
  
          Returns:
              The build output.
          """
          with track_handler(event=AnalyticsEvent.BUILD_PIPELINE):
              self._prepare_if_possible()
  
              compile_args = self._run_args.copy()
              compile_args.pop("prevent_build_reuse", None)
              if config_path:
                  compile_args["config_path"] = config_path
              if step_configurations:
                  compile_args["step_configurations"] = step_configurations
              if settings:
                  compile_args["settings"] = settings
  
              snapshot, _, _ = self._compile(**compile_args)
              pipeline_id = self.register().id
  
              local_repo = code_repository_utils.find_active_code_repository()
              code_repository = build_utils.verify_local_repository_context(
                  snapshot=snapshot, local_repo_context=local_repo
              )
  
              return build_utils.create_pipeline_build(
                  snapshot=snapshot,
                  pipeline_id=pipeline_id,
                  code_repository=code_repository,
              )
  
      def deploy(
          self,
          deployment_name: str,
          timeout: Optional[int] = None,
          *args: Any,
          **kwargs: Any,
      ) -> DeploymentResponse:
          """Deploy the pipeline for online inference.
  
          Args:
              deployment_name: The name to use for the deployment.
              timeout: The maximum time in seconds to wait for the pipeline to be
                  deployed.
              *args: Pipeline entrypoint input arguments.
              **kwargs: Pipeline entrypoint input keyword arguments.
  
          Returns:
              The deployment response.
          """
          self.prepare(*args, **kwargs)
          snapshot = self._create_snapshot(**self._run_args)
  
          stack = Client().active_stack
  
          stack.prepare_pipeline_submission(snapshot=snapshot)
          return stack.deploy_pipeline(
              snapshot=snapshot,
              deployment_name=deployment_name,
              timeout=timeout,
          )
  
      def _create_snapshot(
          self,
          *,
          run_name: Optional[str] = None,
          enable_cache: Optional[bool] = None,
          enable_artifact_metadata: Optional[bool] = None,
          enable_artifact_visualization: Optional[bool] = None,
          enable_step_logs: Optional[bool] = None,
          enable_pipeline_logs: Optional[bool] = None,
          schedule: Optional[Schedule] = None,
          build: Union[str, "UUID", "PipelineBuildBase", None] = None,
          settings: Optional[Mapping[str, "SettingsOrDict"]] = None,
          step_configurations: Optional[
              Mapping[str, "StepConfigurationUpdateOrDict"]
          ] = None,
          extra: Optional[Dict[str, Any]] = None,
          config_path: Optional[str] = None,
          prevent_build_reuse: bool = False,
          skip_schedule_registration: bool = False,
          **snapshot_request_kwargs: Any,
      ) -> PipelineSnapshotResponse:
          """Create a pipeline snapshot.
  
          Args:
              run_name: Name of the pipeline run.
              enable_cache: If caching should be enabled for this pipeline run.
              enable_artifact_metadata: If artifact metadata should be enabled
                  for this pipeline run.
              enable_artifact_visualization: If artifact visualization should be
                  enabled for this pipeline run.
              enable_step_logs: If step logs should be enabled for this pipeline.
              enable_pipeline_logs: If pipeline logs should be enabled for this
                  pipeline run.
              schedule: Optional schedule to use for the run.
              build: Optional build to use for the run.
  