class_docstr: Abstract base class for Patchers for various "DagRunner"s.
  
  These patcher classes "decorate" the `run` function of the DagRunners.
  The patcher can intercept arguments before/after the run() execution,
  and even skip the real run() execution.
  Users of the patcher can read information about the run using `context`
  which is passed as a context manager of patch(). For example,
  the name of the pipeline can be retrieved with
  `context[DagRunnerPatcher.PIPELINE_NAME]`.
  
  Child class should override get_runner_class() to specify which DagRunner
  class they patch. And they can add more actions in before_run()/after_run()
  as they needed.
class_docstr_tok: ["Abstract","base","class","for","Patchers","for","various","``","DagRunner","''","s.","These","patcher","classes","``","decorate","''","the","`","run","`","function","of","the","DagRunners",".","The","patcher","can","intercept","arguments","before\/after","the","run","(",")","execution",",","and","even","skip","the","real","run","(",")","execution",".","Users","of","the","patcher","can","read","information","about","the","run","using","`","context","`","which","is","passed","as","a","context","manager","of","patch","(",")",".","For","example",",","the","name","of","the","pipeline","can","be","retrieved","with","`","context","[","DagRunnerPatcher.PIPELINE_NAME","]","`",".","Child","class","should","override","get_runner_class","(",")","to","specify","which","DagRunner","class","they","patch",".","And","they","can","add","more","actions","in","before_run","(",")","\/after_run","(",")","as","they","needed","."]
docstr: Decorate the run function.
docstr_tok: ["Decorate","the","run","function","."]
code: def _decorate(self, f):
      """Decorate the run function."""
  
      @functools.wraps(f)
      def wrapper(*args, **kwargs):
          # Make sure that the run() is only called exactly once.
          if self._run_called:
              raise RuntimeError(
                  f"{self.get_runner_class().__name__}.run() called more than once."
              )
          self._run_called = True
  
          assert len(args) >= 1, "run() must have self."
          runner = args[0]
          if "pipeline" in kwargs:
              pipeline = kwargs["pipeline"]
          else:
              assert (
                  len(args) >= 2
              ), "run() must have pipeline argument."
              pipeline = args[1]
  
          if isinstance(pipeline, tfx_pipeline.Pipeline):
              self._context[
                  self.PIPELINE_NAME
              ] = pipeline.pipeline_info.pipeline_name
              self._context[
                  self.PIPELINE_ROOT
              ] = pipeline.pipeline_info.pipeline_root
          else:  # pipeline_pb2.Pipeline
              self._context[
                  self.PIPELINE_NAME
              ] = pipeline.pipeline_info.id
              self._context[
                  self.PIPELINE_ROOT
              ] = (
                  pipeline.runtime_spec.pipeline_root.runtime_parameter.default_value.string_value
              )
  
          self._before_run(runner, pipeline, self._context)
          if self._call_real_run:
              result = f(*args, **kwargs)
          else:
              result = None
          self._after_run(runner, pipeline, self._context)
          return result
  
      return wrapper
  
file_path: tfx/tfx/tools/cli/handler/dag_runner_patcher.py
imports: import abc
  import contextlib
  import functools
  from typing import Any, MutableMapping, Type, Union
  from tfx.orchestration import pipeline
  from tfx.orchestration import tfx_runner
  from tfx.orchestration.portable import tfx_runner
  from tfx.proto.orchestration import pipeline_pb2
label_desc: Use image node_id 7 for calling the DagRunnerPatcher obj's underlying member method code with example usage: obj._decorate(f) and returns: wrapper, result