code: def to_pipeline(
      pipeline_name: str,
      step: Callable,
      role: Optional[str] = None,
      transformation_code: Optional[TransformationCode] = None,
      max_retries: Optional[int] = None,
      tags: Optional[List[Tuple[str, str]]] = None,
      sagemaker_session: Optional[Session] = None,
  ) -> str:
      """Creates a sagemaker pipeline that takes in a callable as a training step.
  
      To configure training step used in sagemaker pipeline, input argument step needs to be wrapped
      by remote decorator in module sagemaker.remote_function. If not wrapped by remote decorator,
      default configurations in sagemaker.remote_function.job._JobSettings will be used to create
      training step.
  
      Args:
          pipeline_name (str): The name of the pipeline.
          step (Callable): A user provided function wrapped by feature_processor and optionally
              wrapped by remote_decorator.
          role (Optional[str]): The Amazon Resource Name (ARN) of the role used by the pipeline to
              access and create resources. If not specified, it will default to the credentials
              provided by the AWS configuration chain.
          transformation_code (Optional[str]): The data source for a reference to the transformation
              code for Lineage tracking. This code is not used for actual transformation.
          max_retries (Optional[int]): The number of times to retry sagemaker pipeline step.
              If not specified, sagemaker pipline step will not retry.
          tags (List[Tuple[str, str]): A list of tags attached to the pipeline and all corresponding
              lineage resources that support tags. If not specified, no custom tags will be attached.
          sagemaker_session (Optional[Session]): Session object which manages interactions
              with Amazon SageMaker APIs and any other AWS services needed. If not specified, the
              function creates one using the default AWS configuration chain.
      Returns:
          str: SageMaker Pipeline ARN.
      """
  
      _validate_input_for_to_pipeline_api(pipeline_name, step)
      if tags:
          _validate_tags_for_to_pipeline_api(tags)
  
      _sagemaker_session = sagemaker_session or Session()
  
      _validate_lineage_resources_for_to_pipeline_api(
          step.feature_processor_config, _sagemaker_session
      )
  
      remote_decorator_config = _get_remote_decorator_config_from_input(
          wrapped_func=step, sagemaker_session=_sagemaker_session
      )
      _role = role or get_execution_role(_sagemaker_session)
  
      runtime_env_manager = RuntimeEnvironmentManager()
      client_python_version = (
          runtime_env_manager._current_python_version()
      )
      config_uploader = ConfigUploader(
          remote_decorator_config, runtime_env_manager
      )
  
      s3_base_uri = s3_path_join(
          remote_decorator_config.s3_root_uri, pipeline_name
      )
  
      (
          input_data_config,
          spark_dependency_paths,
      ) = config_uploader.prepare_step_input_channel_for_spark_mode(
          func=getattr(step, "wrapped_func", step),
          s3_base_uri=s3_base_uri,
          sagemaker_session=_sagemaker_session,
      )
  
      estimator_request_dict = (
          _prepare_estimator_request_from_remote_decorator_config(
              remote_decorator_config=remote_decorator_config,
              s3_base_uri=s3_base_uri,
              client_python_version=client_python_version,
              spark_dependency_paths=spark_dependency_paths,
          )
      )
  
      training_step_request_dict = dict(
          name="-".join([pipeline_name, "feature-processor"]),
          estimator=Estimator(**estimator_request_dict),
          inputs=input_data_config,
      )
  
      if max_retries:
          training_step_request_dict["retry_policies"] = [
              StepRetryPolicy(
                  exception_types=[
                      StepExceptionTypeEnum.SERVICE_FAULT,
                      StepExceptionTypeEnum.THROTTLING,
                  ],
                  max_attempts=max_retries,
              ),
              SageMakerJobStepRetryPolicy(
                  exception_types=[
                      SageMakerJobExceptionTypeEnum.INTERNAL_ERROR,
                      SageMakerJobExceptionTypeEnum.CAPACITY_ERROR,
                      SageMakerJobExceptionTypeEnum.RESOURCE_LIMIT,
                  ],
                  max_attempts=max_retries,
              ),
          ]
  
      pipeline_request_dict = dict(
          name=pipeline_name,
          steps=[TrainingStep(**training_step_request_dict)],
          sagemaker_session=_sagemaker_session,
          parameters=[SCHEDULED_TIME_PIPELINE_PARAMETER],
      )
      pipeline_tags = [
          dict(
              Key=FEATURE_PROCESSOR_TAG_KEY,
              Value=FEATURE_PROCESSOR_TAG_VALUE,
          )
      ]
      if tags:
          pipeline_tags.extend([dict(Key=k, Value=v) for k, v in tags])
  
      pipeline = Pipeline(**pipeline_request_dict)
      logger.info(
          "Creating/Updating sagemaker pipeline %s", pipeline_name
      )
      pipeline.upsert(
          role_arn=_role,
          tags=pipeline_tags,
      )
      logger.info("Created sagemaker pipeline %s", pipeline_name)
  
      describe_pipeline_response = pipeline.describe()
      pipeline_arn = describe_pipeline_response["PipelineArn"]
      tags_propagate_to_lineage_resources = (
          _get_tags_from_pipeline_to_propagate_to_lineage_resources(
              pipeline_arn, _sagemaker_session
          )
      )
  
      lineage_handler = FeatureProcessorLineageHandler(
          pipeline_name=pipeline_name,
          pipeline_arn=pipeline_arn,
          pipeline=describe_pipeline_response,
          inputs=_get_feature_processor_inputs(wrapped_func=step),
          output=_get_feature_processor_outputs(wrapped_func=step),
          transformation_code=transformation_code,
          sagemaker_session=_sagemaker_session,
      )
      lineage_handler.create_lineage(
          tags_propagate_to_lineage_resources
      )
      lineage_handler.upsert_tags_for_lineage_resources(
          tags_propagate_to_lineage_resources
      )
  
      pipeline_lineage_names: Dict[
          str, str
      ] = lineage_handler.get_pipeline_lineage_names()
  
      if pipeline_lineage_names is None:
          raise RuntimeError(
              "Failed to retrieve pipeline lineage. Pipeline Lineage does not exist"
          )
  
      pipeline.upsert(
          role_arn=_role,
          tags=[
              {
                  "Key": PIPELINE_CONTEXT_NAME_TAG_KEY,
                  "Value": pipeline_lineage_names[
                      "pipeline_context_name"
                  ],
              },
              {
                  "Key": PIPELINE_VERSION_CONTEXT_NAME_TAG_KEY,
                  "Value": pipeline_lineage_names[
                      "pipeline_version_context_name"
                  ],
              },
          ],
      )
      return pipeline_arn
  
imports: from __future__ import absolute_import
  import logging
  import json
  import re
  from datetime import datetime
  from typing import Callable, List, Optional, Dict, Sequence, Union, Any, Tuple
  import pytz
  from botocore.exceptions import ClientError
  from sagemaker.feature_store.feature_processor._config_uploader import ConfigUploader
  from sagemaker.feature_store.feature_processor._enums import FeatureProcessorMode
  from sagemaker.feature_store.feature_processor._event_bridge_rule_helper import EventBridgeRuleHelper
  from sagemaker.feature_store.feature_processor._feature_processor_pipeline_events import FeatureProcessorPipelineEvents
  from sagemaker.feature_store.feature_processor.lineage._feature_processor_lineage_name_helper import _get_feature_group_lineage_context_name, _get_feature_group_pipeline_lineage_context_name, _get_feature_group_pipeline_version_lineage_context_name, _get_feature_processor_pipeline_lineage_context_name, _get_feature_processor_pipeline_version_lineage_context_name
  from sagemaker.lineage import context
  from sagemaker.lineage._utils import get_resource_name_from_arn
  from sagemaker.remote_function.runtime_environment.runtime_environment_manager import RuntimeEnvironmentManager
  from sagemaker.remote_function.spark_config import SparkConfig
  from sagemaker.vpc_utils import SUBNETS_KEY, SECURITY_GROUP_IDS_KEY
  from sagemaker.feature_store.feature_processor._constants import EXECUTION_TIME_PIPELINE_PARAMETER, EXECUTION_TIME_PIPELINE_PARAMETER_FORMAT, RESOURCE_NOT_FOUND_EXCEPTION, SPARK_JAR_FILES_PATH, SPARK_PY_FILES_PATH, SPARK_FILES_PATH, FEATURE_PROCESSOR_TAG_KEY, FEATURE_PROCESSOR_TAG_VALUE, PIPELINE_CONTEXT_TYPE, DEFAULT_SCHEDULE_STATE, SCHEDULED_TIME_PIPELINE_PARAMETER, PIPELINE_CONTEXT_NAME_TAG_KEY, PIPELINE_VERSION_CONTEXT_NAME_TAG_KEY, PIPELINE_NAME_MAXIMUM_LENGTH, RESOURCE_NOT_FOUND, FEATURE_GROUP_ARN_REGEX_PATTERN, TO_PIPELINE_RESERVED_TAG_KEYS, DEFAULT_TRIGGER_STATE, EVENTBRIDGE_RULE_ARN_REGEX_PATTERN
  from sagemaker.feature_store.feature_processor._feature_processor_config import FeatureProcessorConfig
  from sagemaker.s3 import s3_path_join
  from sagemaker import Session, get_execution_role
  from sagemaker.feature_store.feature_processor._event_bridge_scheduler_helper import EventBridgeSchedulerHelper
  from sagemaker.workflow.pipeline import Pipeline
  from sagemaker.workflow.retry import StepRetryPolicy, StepExceptionTypeEnum, SageMakerJobStepRetryPolicy, SageMakerJobExceptionTypeEnum
  from sagemaker.workflow.steps import TrainingStep
  from sagemaker.estimator import Estimator
  from sagemaker.feature_store.feature_processor.lineage._feature_processor_lineage import FeatureProcessorLineageHandler, TransformationCode
  from sagemaker.remote_function.job import _JobSettings, JOBS_CONTAINER_ENTRYPOINT, SPARK_APP_SCRIPT_PATH
  from sagemaker.feature_store.feature_processor._data_source import CSVDataSource, FeatureGroupDataSource, ParquetDataSource
label_desc: Use image node_id 1 for calling a global function with example usage: to_pipeline(pipeline_name, step, role, transformation_code, max_retries, tags, sagemaker_session) and returns: pipeline_arn