class_docstr: Executor for Slack component.
class_docstr_tok: ["Executor","for","Slack","component","."]
docstr: Get human review result on a model through Slack channel.
  
  Args:
    input_dict: Input dict from input key to a list of artifacts, including:
      - model_export: exported model from trainer.
      - model_blessing: model blessing path from evaluator.
    output_dict: Output dict from key to a list of artifacts, including:
      - slack_blessing: model blessing result.
    exec_properties: A dict of execution properties, including:
      - slack_token: Token used to setup connection with slack server.
      - slack_channel_id: The id of the Slack channel to send and receive
        messages.
      - timeout_sec: How long do we wait for response, in seconds.
  
  Returns:
    None
  
  Raises:
    TimeoutError:
      When there is no decision made within timeout_sec.
    ConnectionError:
      When connection to slack server cannot be established.
docstr_tok: ["Get","human","review","result","on","a","model","through","Slack","channel",".","Args",":","input_dict",":","Input","dict","from","input","key","to","a","list","of","artifacts",",","including",":","-","model_export",":","exported","model","from","trainer",".","-","model_blessing",":","model","blessing","path","from","evaluator",".","output_dict",":","Output","dict","from","key","to","a","list","of","artifacts",",","including",":","-","slack_blessing",":","model","blessing","result",".","exec_properties",":","A","dict","of","execution","properties",",","including",":","-","slack_token",":","Token","used","to","setup","connection","with","slack","server",".","-","slack_channel_id",":","The","id","of","the","Slack","channel","to","send","and","receive","messages",".","-","timeout_sec",":","How","long","do","we","wait","for","response",",","in","seconds",".","Returns",":","None","Raises",":","TimeoutError",":","When","there","is","no","decision","made","within","timeout_sec",".","ConnectionError",":","When","connection","to","slack","server","can","not","be","established","."]
code: def Do(
      self,
      input_dict: Dict[str, List[types.Artifact]],
      output_dict: Dict[str, List[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      """Get human review result on a model through Slack channel.
  
      Args:
        input_dict: Input dict from input key to a list of artifacts, including:
          - model_export: exported model from trainer.
          - model_blessing: model blessing path from evaluator.
        output_dict: Output dict from key to a list of artifacts, including:
          - slack_blessing: model blessing result.
        exec_properties: A dict of execution properties, including:
          - slack_token: Token used to setup connection with slack server.
          - slack_channel_id: The id of the Slack channel to send and receive
            messages.
          - timeout_sec: How long do we wait for response, in seconds.
  
      Returns:
        None
  
      Raises:
        TimeoutError:
          When there is no decision made within timeout_sec.
        ConnectionError:
          When connection to slack server cannot be established.
  
      """
      self._log_startup(input_dict, output_dict, exec_properties)
  
      # Fetch execution properties from exec_properties dict.
      slack_token = exec_properties["slack_token"]
      slack_channel_id = exec_properties["slack_channel_id"]
      timeout_sec = exec_properties["timeout_sec"]
  
      # Fetch input URIs from input_dict.
      model_export_uri = artifact_utils.get_single_uri(
          input_dict["model"]
      )
      model_blessing = artifact_utils.get_single_instance(
          input_dict["model_blessing"]
      )
  
      # Fetch output artifact from output_dict.
      slack_blessing = artifact_utils.get_single_instance(
          output_dict["slack_blessing"]
      )
  
      # We only consider a model as blessed if both of the following conditions
      # are met:
      # - The model is blessed by evaluator. This is determined by looking
      #   for file named 'BLESSED' from the output from Evaluator.
      # - The model is blessed by a human reviewer. This logic is in
      #   _fetch_slack_blessing().
      slack_response = None
      with Timeout(timeout_sec):
          if model_utils.is_model_blessed(model_blessing):
              slack_response = self._fetch_slack_blessing(
                  slack_token, slack_channel_id, model_export_uri
              )
  
      # If model is blessed, write an empty file named 'BLESSED' in the assigned
      # output path. Otherwise, write an empty file named 'NOT_BLESSED' instead.
      if slack_response and slack_response.approved:
          io_utils.write_string_file(
              os.path.join(slack_blessing.uri, "BLESSED"), ""
          )
          slack_blessing.set_int_custom_property("blessed", 1)
      else:
          io_utils.write_string_file(
              os.path.join(slack_blessing.uri, "NOT_BLESSED"), ""
          )
          slack_blessing.set_int_custom_property("blessed", 0)
      if slack_response:
          slack_blessing.set_string_custom_property(
              "slack_decision_maker", slack_response.user_id
          )
          slack_blessing.set_string_custom_property(
              "slack_decision_message", slack_response.message
          )
          slack_blessing.set_string_custom_property(
              "slack_decision_channel", slack_response.slack_channel_id
          )
          slack_blessing.set_string_custom_property(
              "slack_decision_thread", slack_response.thread_ts
          )
      absl.logging.info(
          "Blessing result written to %s.", slack_blessing.uri
      )
  
imports: import os
  import signal
  from typing import Any, Dict, List
  import absl
  import attr
  import slack
  from tfx import types
  from tfx.components.util import model_utils
  from tfx.dsl.components.base import base_executor
  from tfx.types import artifact_utils
  from tfx.utils import io_utils
label_desc: Use image node_id 2 for calling the Executor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types