class_docstr: TFX Pusher executor to push the new TF model to a filesystem target.
  
  The Pusher component is used to deploy a validated model to a filesystem
  target or serving environment using tf.serving.  Pusher depends on the outputs
  of ModelValidator to determine if a model is ready to push. A model is
  considered to be safe to push only if ModelValidator has marked it as BLESSED.
  A push action delivers the model exports produced by Trainer to the
  destination defined in the ``push_destination`` of the component config.
  
  To include Pusher in a TFX pipeline, configure your pipeline similar to
  https://github.com/tensorflow/tfx/blob/master/tfx/examples/chicago_taxi_pipeline/taxi_pipeline_simple.py#L104.
  
  For more details on tf.serving itself, please refer to
  https://tensorflow.org/tfx/guide/pusher.  For a tutuorial on TF Serving,
  please refer to https://www.tensorflow.org/tfx/guide/serving.
class_docstr_tok: ["TFX","Pusher","executor","to","push","the","new","TF","model","to","a","filesystem","target",".","The","Pusher","component","is","used","to","deploy","a","validated","model","to","a","filesystem","target","or","serving","environment","using","tf.serving",".","Pusher","depends","on","the","outputs","of","ModelValidator","to","determine","if","a","model","is","ready","to","push",".","A","model","is","considered","to","be","safe","to","push","only","if","ModelValidator","has","marked","it","as","BLESSED",".","A","push","action","delivers","the","model","exports","produced","by","Trainer","to","the","destination","defined","in","the","``","push_destination","``","of","the","component","config",".","To","include","Pusher","in","a","TFX","pipeline",",","configure","your","pipeline","similar","to","https",":","\/\/github.com\/tensorflow\/tfx\/blob\/master\/tfx\/examples\/chicago_taxi_pipeline\/taxi_pipeline_simple.py","#","L104",".","For","more","details","on","tf.serving","itself",",","please","refer","to","https",":","\/\/tensorflow.org\/tfx\/guide\/pusher",".","For","a","tutuorial","on","TF","Serving",",","please","refer","to","https",":","\/\/www.tensorflow.org\/tfx\/guide\/serving","."]
docstr: Push model to target directory if blessed.
  
  Args:
    input_dict: Input dict from input key to a list of artifacts, including:
      - model: exported model from trainer.
      - model_blessing: model blessing path from model_validator.  A push
        action delivers the model exports produced by Trainer to the
        destination defined in component config.
    output_dict: Output dict from key to a list of artifacts, including:
      - pushed_model: A list of 'ModelPushPath' artifact of size one. It will
        include the model in this push execution if the model was pushed.
    exec_properties: A dict of execution properties, including:
      - push_destination: JSON string of pusher_pb2.PushDestination instance,
        providing instruction of destination to push model.
  
  Returns:
    None
docstr_tok: ["Push","model","to","target","directory","if","blessed",".","Args",":","input_dict",":","Input","dict","from","input","key","to","a","list","of","artifacts",",","including",":","-","model",":","exported","model","from","trainer",".","-","model_blessing",":","model","blessing","path","from","model_validator",".","A","push","action","delivers","the","model","exports","produced","by","Trainer","to","the","destination","defined","in","component","config",".","output_dict",":","Output","dict","from","key","to","a","list","of","artifacts",",","including",":","-","pushed_model",":","A","list","of","'ModelPushPath","'","artifact","of","size","one",".","It","will","include","the","model","in","this","push","execution","if","the","model","was","pushed",".","exec_properties",":","A","dict","of","execution","properties",",","including",":","-","push_destination",":","JSON","string","of","pusher_pb2.PushDestination","instance",",","providing","instruction","of","destination","to","push","model",".","Returns",":","None"]
code: def Do(
      self,
      input_dict: Dict[str, List[types.Artifact]],
      output_dict: Dict[str, List[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      """Push model to target directory if blessed.
  
      Args:
        input_dict: Input dict from input key to a list of artifacts, including:
          - model: exported model from trainer.
          - model_blessing: model blessing path from model_validator.  A push
            action delivers the model exports produced by Trainer to the
            destination defined in component config.
        output_dict: Output dict from key to a list of artifacts, including:
          - pushed_model: A list of 'ModelPushPath' artifact of size one. It will
            include the model in this push execution if the model was pushed.
        exec_properties: A dict of execution properties, including:
          - push_destination: JSON string of pusher_pb2.PushDestination instance,
            providing instruction of destination to push model.
  
      Returns:
        None
      """
      self._log_startup(input_dict, output_dict, exec_properties)
      model_push = artifact_utils.get_single_instance(
          output_dict[standard_component_specs.PUSHED_MODEL_KEY]
      )
      if not self.CheckBlessing(input_dict):
          self._MarkNotPushed(model_push)
          return
      model_path = self.GetModelPath(input_dict)
  
      # Push model to the destination, which can be listened by a model server.
      #
      # If model is already successfully copied to outside before, stop copying.
      # This is because model validator might blessed same model twice (check
      # mv driver) with different blessing output, we still want Pusher to
      # handle the mv output again to keep metadata tracking, but no need to
      # copy to outside path again..
      # TODO(jyzhao): support rpc push and verification.
      push_destination = pusher_pb2.PushDestination()
      proto_utils.json_to_proto(
          exec_properties[
              standard_component_specs.PUSH_DESTINATION_KEY
          ],
          push_destination,
      )
  
      destination_kind = push_destination.WhichOneof("destination")
      if destination_kind == "filesystem":
          fs_config = push_destination.filesystem
          if fs_config.versioning == _Versioning.AUTO:
              fs_config.versioning = _Versioning.UNIX_TIMESTAMP
          if fs_config.versioning == _Versioning.UNIX_TIMESTAMP:
              model_version = str(int(time.time()))
          else:
              raise NotImplementedError(
                  "Invalid Versioning {}".format(fs_config.versioning)
              )
          logging.info("Model version: %s", model_version)
          serving_path = os.path.join(
              fs_config.base_directory, model_version
          )
  
          if fileio.exists(serving_path):
              logging.info(
                  "Destination directory %s already exists, skipping current push.",
                  serving_path,
              )
          else:
              # For TensorFlow SavedModel, saved_model.pb file should be the last file
              # to be copied as TF serving and other codes rely on that file as an
              # indication that the model is available.
              # https://github.com/tensorflow/tensorflow/blob/d5b3c79b4804134d0d17bfce9f312151f6337dba/tensorflow/python/saved_model/save.py#L1445
              io_utils.copy_dir(
                  model_path,
                  serving_path,
                  deny_regex_patterns=[r"saved_model\.pb"],
              )
              saved_model_path = os.path.join(
                  model_path, "saved_model.pb"
              )
              if fileio.exists(saved_model_path):
                  io_utils.copy_file(
                      saved_model_path,
                      os.path.join(serving_path, "saved_model.pb"),
                  )
              logging.info(
                  "Model written to serving path %s.", serving_path
              )
      else:
          raise NotImplementedError(
              "Invalid push destination {}".format(destination_kind)
          )
  
      # Copy the model to pushing uri for archiving.
      io_utils.copy_dir(model_path, model_push.uri)
      self._MarkPushed(
          model_push,
          pushed_destination=serving_path,
          pushed_version=model_version,
      )
      logging.info("Model pushed to %s.", model_push.uri)
  
imports: import os
  import time
  from typing import Any, Dict, List, Optional
  from absl import logging
  from tfx import types
  from tfx.components.util import model_utils
  from tfx.dsl.components.base import base_executor
  from tfx.dsl.io import fileio
  from tfx.proto import pusher_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_component_specs
  from tfx.utils import io_utils
  from tfx.utils import path_utils
  from tfx.utils import proto_utils
label_desc: Use image node_id 3 for calling the Executor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types