class_docstr: MetaflowTask prepares a Flow instance for execution of a single step.
class_docstr_tok: ["MetaflowTask","prepares","a","Flow","instance","for","execution","of","a","single","step","."]
code: def run_step(
      self,
      step_name,
      run_id,
      task_id,
      origin_run_id,
      input_paths,
      split_index,
      retry_count,
      max_user_code_retries,
  ):
      if run_id and task_id:
          self.metadata.register_run_id(run_id)
          self.metadata.register_task_id(
              run_id, step_name, task_id, retry_count
          )
      else:
          raise MetaflowInternalError(
              "task.run_step needs a valid run_id and task_id"
          )
  
      if retry_count >= MAX_ATTEMPTS:
          # any results with an attempt ID >= MAX_ATTEMPTS will be ignored
          # by datastore, so running a task with such a retry_could would
          # be pointless and dangerous
          raise MetaflowInternalError(
              "Too many task attempts (%d)! MAX_ATTEMPTS exceeded."
              % retry_count
          )
  
      metadata_tags = ["attempt_id:{0}".format(retry_count)]
  
      metadata = [
          MetaDatum(
              field="attempt",
              value=str(retry_count),
              type="attempt",
              tags=metadata_tags,
          ),
          MetaDatum(
              field="origin-run-id",
              value=str(origin_run_id),
              type="origin-run-id",
              tags=metadata_tags,
          ),
          MetaDatum(
              field="ds-type",
              value=self.flow_datastore.TYPE,
              type="ds-type",
              tags=metadata_tags,
          ),
          MetaDatum(
              field="ds-root",
              value=self.flow_datastore.datastore_root,
              type="ds-root",
              tags=metadata_tags,
          ),
      ]
      trace_id = get_trace_id()
      if trace_id:
          metadata.append(
              MetaDatum(
                  field="otel-trace-id",
                  value=trace_id,
                  type="trace-id",
                  tags=metadata_tags,
              )
          )
  
      self.metadata.register_metadata(
          run_id,
          step_name,
          task_id,
          metadata,
      )
  
      step_func = getattr(self.flow, step_name)
      decorators = step_func.decorators
  
      node = self.flow._graph[step_name]
      join_type = None
      if node.type == "join":
          join_type = self.flow._graph[node.split_parents[-1]].type
  
      # 1. initialize output datastore
      output = self.flow_datastore.get_task_datastore(
          run_id, step_name, task_id, attempt=retry_count, mode="w"
      )
  
      output.init_task()
  
      if input_paths:
          # 2. initialize input datastores
          inputs = self._init_data(run_id, join_type, input_paths)
  
          # 3. initialize foreach state
          self._init_foreach(step_name, join_type, inputs, split_index)
  
      # 4. initialize the current singleton
      current._set_env(
          flow=self.flow,
          run_id=run_id,
          step_name=step_name,
          task_id=task_id,
          retry_count=retry_count,
          origin_run_id=origin_run_id,
          namespace=resolve_identity(),
          username=get_username(),
          metadata_str="%s@%s"
          % (
              self.metadata.__class__.TYPE,
              self.metadata.__class__.INFO,
          ),
          is_running=True,
          tags=self.metadata.sticky_tags,
      )
  
      # 5. run task
      output.save_metadata(
          {
              "task_begin": {
                  "code_package_sha": os.environ.get(
                      "METAFLOW_CODE_SHA"
                  ),
                  "code_package_ds": os.environ.get("METAFLOW_CODE_DS"),
                  "code_package_url": os.environ.get(
                      "METAFLOW_CODE_URL"
                  ),
                  "retry_count": retry_count,
              }
          }
      )
      logger = self.event_logger
      start = time.time()
      self.metadata.start_task_heartbeat(
          self.flow.name, run_id, step_name, task_id
      )
      try:
          msg = {
              "task_id": task_id,
              "msg": "task starting",
              "step_name": step_name,
              "run_id": run_id,
              "flow_name": self.flow.name,
              "ts": round(time.time()),
          }
          logger.log(msg)
  
          self.flow._current_step = step_name
          self.flow._success = False
          self.flow._task_ok = None
          self.flow._exception = None
          # Note: All internal flow attributes (ie: non-user artifacts)
          # should either be set prior to running the user code or listed in
          # FlowSpec._EPHEMERAL to allow for proper merging/importing of
          # user artifacts in the user's step code.
  
          if join_type:
              # Join step:
  
              # Ensure that we have the right number of inputs. The
              # foreach case is checked above.
              if join_type != "foreach" and len(inputs) != len(
                  node.in_funcs
              ):
                  raise MetaflowDataMissing(
                      "Join *%s* expected %d "
                      "inputs but only %d inputs "
                      "were found"
                      % (step_name, len(node.in_funcs), len(inputs))
                  )
  
              # Multiple input contexts are passed in as an argument
              # to the step function.
              input_obj = Inputs(
                  self._clone_flow(inp) for inp in inputs
              )
              self.flow._set_datastore(output)
              # initialize parameters (if they exist)
              # We take Parameter values from the first input,
              # which is always safe since parameters are read-only
              current._update_env(
                  {
                      "parameter_names": self._init_parameters(
                          inputs[0], passdown=True
                      )
                  }
              )
          else:
              # Linear step:
              # We are running with a single input context.
              # The context is embedded in the flow.
              if len(inputs) > 1:
                  # This should be captured by static checking but
                  # let's assert this again
                  raise MetaflowInternalError(
                      "Step *%s* is not a join "
                      "step but it gets multiple "
                      "inputs." % step_name
                  )
              self.flow._set_datastore(inputs[0])
              if input_paths:
                  # initialize parameters (if they exist)
                  # We take Parameter values from the first input,
                  # which is always safe since parameters are read-only
                  current._update_env(
                      {
                          "parameter_names": self._init_parameters(
                              inputs[0], passdown=False
                          )
                      }
                  )
  
          for deco in decorators:
              deco.task_pre_step(
                  step_name,
                  output,
                  self.metadata,
                  run_id,
                  task_id,
                  self.flow,
                  self.flow._graph,
                  retry_count,
                  max_user_code_retries,
                  self.ubf_context,
                  inputs,
              )
  
          for deco in decorators:
              # decorators can actually decorate the step function,
              # or they can replace it altogether. This functionality
              # is used e.g. by catch_decorator which switches to a
              # fallback code if the user code has failed too many
              # times.
              step_func = deco.task_decorate(
                  step_func,
                  self.flow,
                  self.flow._graph,
                  retry_count,
                  max_user_code_retries,
                  self.ubf_context,
              )
  
          if join_type:
              self._exec_step_function(step_func, input_obj)
          else:
              self._exec_step_function(step_func)
  
          for deco in decorators:
              deco.task_post_step(
                  step_name,
                  self.flow,
                  self.flow._graph,
                  retry_count,
                  max_user_code_retries,
              )
  
          self.flow._task_ok = True
          self.flow._success = True
  
      except Exception as ex:
          tsk_msg = {
              "task_id": task_id,
              "exception_msg": str(ex),
              "msg": "task failed with exception",
              "step_name": step_name,
              "run_id": run_id,
              "flow_name": self.flow.name,
          }
          logger.log(tsk_msg)
  
          exception_handled = False
          for deco in decorators:
              res = deco.task_exception(
                  ex,
                  step_name,
                  self.flow,
                  self.flow._graph,
                  retry_count,
                  max_user_code_retries,
              )
              exception_handled = bool(res) or exception_handled
  
          if exception_handled:
              self.flow._task_ok = True
          else:
              self.flow._task_ok = False
              self.flow._exception = MetaflowExceptionWrapper(ex)
              print("%s failed:" % self.flow, file=sys.stderr)
              raise
  
      finally:
          if self.ubf_context == UBF_CONTROL:
              self._finalize_control_task()
  
          end = time.time() - start
  
          msg = {
              "task_id": task_id,
              "msg": "task ending",
              "step_name": step_name,
              "run_id": run_id,
              "flow_name": self.flow.name,
              "ts": round(time.time()),
              "runtime": round(end),
          }
          logger.log(msg)
  
          attempt_ok = str(bool(self.flow._task_ok))
          self.metadata.register_metadata(
              run_id,
              step_name,
              task_id,
              [
                  MetaDatum(
                      field="attempt_ok",
                      value=attempt_ok,
                      type="internal_attempt_status",
                      tags=["attempt_id:{0}".format(retry_count)],
                  )
              ],
          )
  
          output.save_metadata({"task_end": {}})
          output.persist(self.flow)
  
          # this writes a success marker indicating that the
          # "transaction" is done
          output.done()
  
          # final decorator hook: The task results are now
          # queryable through the client API / datastore
          for deco in decorators:
              deco.task_finished(
                  step_name,
                  self.flow,
                  self.flow._graph,
                  self.flow._task_ok,
                  retry_count,
                  max_user_code_retries,
              )
  
          # terminate side cars
          self.metadata.stop_heartbeat()
  
imports: from __future__ import print_function
  from io import BytesIO
  import math
  import sys
  import os
  import time
  from types import MethodType, FunctionType
  from metaflow.datastore.exceptions import DataException
  from .metaflow_config import MAX_ATTEMPTS
  from .metadata import MetaDatum
  from .mflog import TASK_LOG_SOURCE
  from .datastore import Inputs, TaskDataStoreSet
  from .exception import MetaflowInternalError, MetaflowDataMissing, MetaflowExceptionWrapper
  from .unbounded_foreach import UBF_CONTROL
  from .util import all_equal, get_username, resolve_identity, unicode_type
  from .current import current
  from metaflow.tracing import get_trace_id
  from collections import namedtuple
label_desc: Use image node_id 9 for calling the MetaflowTask obj's underlying member method code with example usage: obj.run_step(step_name, run_id, task_id, origin_run_id, input_paths, split_index, retry_count, max_user_code_retries) without return types