class_docstr: Class to Create and Update FeatureProcessor Lineage Entities.
  
  Attributes:
      pipeline_name (str): Pipeline Name.
      pipeline_arn (str): The ARN of the Pipeline.
      pipeline (str): The details of the Pipeline.
      inputs (Sequence[Union[FeatureGroupDataSource, CSVDataSource, ParquetDataSource,
          BaseDataSource]]): The inputs to the Feature processor.
      output (str): The output Feature Group.
      transformation_code (TransformationCode): The Transformation Code for Feature Processor.
      sagemaker_session (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.
class_docstr_tok: ["Class","to","Create","and","Update","FeatureProcessor","Lineage","Entities",".","Attributes",":","pipeline_name","(","str",")",":","Pipeline","Name",".","pipeline_arn","(","str",")",":","The","ARN","of","the","Pipeline",".","pipeline","(","str",")",":","The","details","of","the","Pipeline",".","inputs","(","Sequence","[","Union","[","FeatureGroupDataSource",",","CSVDataSource",",","ParquetDataSource",",","BaseDataSource","]","]",")",":","The","inputs","to","the","Feature","processor",".","output","(","str",")",":","The","output","Feature","Group",".","transformation_code","(","TransformationCode",")",":","The","Transformation","Code","for","Feature","Processor",".","sagemaker_session","(","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","."]
docstr: Add or update tags for lineage resources using tags attached to sagemaker pipeline as
  
  source of truth.
  
  Args:
      tags (List[Dict[str, str]]): Custom tags to be attached to lineage resources.
code: def upsert_tags_for_lineage_resources(
      self, tags: List[Dict[str, str]]
  ) -> None:
      """Add or update tags for lineage resources using tags attached to sagemaker pipeline as
  
      source of truth.
  
      Args:
          tags (List[Dict[str, str]]): Custom tags to be attached to lineage resources.
      """
      if not tags:
          return
      pipeline_context: Context = self._get_pipeline_context()
      current_pipeline_version_context: Context = (
          self._get_pipeline_version_context(
              last_update_time=pipeline_context.properties[
                  LAST_UPDATE_TIME
              ]
          )
      )
      input_raw_data_artifacts: List[
          Artifact
      ] = self._retrieve_input_raw_data_artifacts()
      pipeline_context.set_tags(tags)
      current_pipeline_version_context.set_tags(tags)
      for input_raw_data_artifact in input_raw_data_artifacts:
          input_raw_data_artifact.set_tags(tags)
  
      event_bridge_scheduler_helper = EventBridgeSchedulerHelper(
          self.sagemaker_session,
          self.sagemaker_session.boto_session.client("scheduler"),
      )
      event_bridge_schedule = (
          event_bridge_scheduler_helper.describe_schedule(
              self.pipeline_name
          )
      )
  
      event_bridge_rule_helper = EventBridgeRuleHelper(
          self.sagemaker_session,
          self.sagemaker_session.boto_session.client("events"),
      )
      event_bridge_rule = event_bridge_rule_helper.describe_rule(
          self.pipeline_name
      )
  
      if event_bridge_schedule:
          schedule_artifact_summary = (
              S3LineageEntityHandler._load_artifact_from_s3_uri(
                  s3_uri=event_bridge_schedule["Arn"],
                  sagemaker_session=self.sagemaker_session,
              )
          )
          if schedule_artifact_summary is not None:
              pipeline_schedule_artifact: Artifact = S3LineageEntityHandler.load_artifact_from_arn(
                  artifact_arn=schedule_artifact_summary.artifact_arn,
                  sagemaker_session=self.sagemaker_session,
              )
              pipeline_schedule_artifact.set_tags(tags)
  
      if event_bridge_rule:
          rule_artifact_summary = (
              S3LineageEntityHandler._load_artifact_from_s3_uri(
                  s3_uri=event_bridge_rule["Arn"],
                  sagemaker_session=self.sagemaker_session,
              )
          )
          if rule_artifact_summary:
              pipeline_trigger_artifact: Artifact = (
                  S3LineageEntityHandler.load_artifact_from_arn(
                      artifact_arn=rule_artifact_summary.artifact_arn,
                      sagemaker_session=self.sagemaker_session,
                  )
              )
              pipeline_trigger_artifact.set_tags(tags)
  
imports: from __future__ import absolute_import
  import logging
  from datetime import datetime
  from typing import Optional, Iterator, List, Dict, Set, Sequence, Union
  import attr
  from botocore.exceptions import ClientError
  from sagemaker.feature_store.feature_processor._event_bridge_rule_helper import EventBridgeRuleHelper
  from sagemaker.feature_store.feature_processor._event_bridge_scheduler_helper import EventBridgeSchedulerHelper
  from sagemaker.feature_store.feature_processor.lineage._lineage_association_handler import LineageAssociationHandler
  from sagemaker.feature_store.feature_processor.lineage._feature_group_lineage_entity_handler import FeatureGroupLineageEntityHandler
  from sagemaker.feature_store.feature_processor.lineage._feature_group_contexts import FeatureGroupContexts
  from sagemaker.feature_store.feature_processor.lineage._pipeline_lineage_entity_handler import PipelineLineageEntityHandler
  from sagemaker.feature_store.feature_processor.lineage._pipeline_schedule import PipelineSchedule
  from sagemaker.feature_store.feature_processor.lineage._pipeline_trigger import PipelineTrigger
  from sagemaker.feature_store.feature_processor.lineage._pipeline_version_lineage_entity_handler import PipelineVersionLineageEntityHandler
  from sagemaker.feature_store.feature_processor.lineage._s3_lineage_entity_handler import S3LineageEntityHandler
  from sagemaker.feature_store.feature_processor.lineage._transformation_code import TransformationCode
  from sagemaker.feature_store.feature_processor.lineage.constants import SAGEMAKER, LAST_UPDATE_TIME, PIPELINE_CONTEXT_NAME_KEY, PIPELINE_CONTEXT_VERSION_NAME_KEY, FEATURE_GROUP_PIPELINE_VERSION_CONTEXT_TYPE, DATA_SET, TRANSFORMATION_CODE, CREATION_TIME, RESOURCE_NOT_FOUND, ERROR, CODE, LAST_MODIFIED_TIME, TRANSFORMATION_CODE_STATUS_INACTIVE, TRANSFORMATION_CODE_STATUS_ACTIVE, CONTRIBUTED_TO
  from sagemaker.lineage.context import Context
  from sagemaker.lineage.artifact import Artifact
  from sagemaker.lineage.association import AssociationSummary
  from sagemaker import Session
  from sagemaker.feature_store.feature_processor._data_source import CSVDataSource, FeatureGroupDataSource, ParquetDataSource, BaseDataSource
label_desc: Use image node_id 5 for calling the FeatureProcessorLineageHandler obj's underlying member method code with example usage: obj.upsert_tags_for_lineage_resources(tags) without return types