class_docstr: Specifies the flow(s) that this flow depends on.
  
  ```
  @trigger_on_finish(flow='FooFlow')
  ```
  or
  ```
  @trigger_on_finish(flows=['FooFlow', 'BarFlow'])
  ```
  This decorator respects the @project decorator and triggers the flow
  when upstream runs within the same namespace complete successfully
  
  Additionally, you can specify project aware upstream flow dependencies
  by specifying the fully qualified project_flow_name.
  ```
  @trigger_on_finish(flow='my_project.branch.my_branch.FooFlow')
  ```
  or
  ```
  @trigger_on_finish(flows=['my_project.branch.my_branch.FooFlow', 'BarFlow'])
  ```
  
  You can also specify just the project or project branch (other values will be
  inferred from the current project or project branch):
  ```
  @trigger_on_finish(flow={"name": "FooFlow", "project": "my_project", "project_branch": "branch"})
  ```
  
  Note that `branch` is typically one of:
    - `prod`
    - `user.bob`
    - `test.my_experiment`
    - `prod.staging`
  
  Parameters
  ----------
  flow : Union[str, Dict[str, str]], optional
      Upstream flow dependency for this flow.
  flows : List[Union[str, Dict[str, str]], optional
      Upstream flow dependencies for this flow.
  options : dict, optional
      Backend-specific configuration for tuning eventing behavior.
code: def flow_init(
      self,
      flow_name,
      graph,
      environment,
      flow_datastore,
      metadata,
      logger,
      echo,
      options,
  ):
      self.triggers = []
      if (
          sum(
              map(
                  bool,
                  (self.attributes["flow"], self.attributes["flows"]),
              )
          )
          > 1
      ):
          raise MetaflowException(
              "Specify only one of *flow* or *flows* "
              "attributes in *@trigger_on_finish* decorator."
          )
      elif self.attributes["flow"]:
          # flow supports the format @trigger_on_finish(flow='FooFlow')
          if is_stringish(self.attributes["flow"]):
              self.triggers.append(
                  {
                      "fq_name": self.attributes["flow"],
                  }
              )
          elif isinstance(self.attributes["flow"], dict):
              if "name" not in self.attributes["flow"]:
                  raise MetaflowException(
                      "The *flow* attribute for *@trigger_on_finish* is missing the "
                      "*name* key."
                  )
              flow_name = self.attributes["flow"]["name"]
  
              if not is_stringish(flow_name) or "." in flow_name:
                  raise MetaflowException(
                      "The *name* attribute of the *flow* is not a valid string"
                  )
              result = {"fq_name": flow_name}
              if "project" in self.attributes["flow"]:
                  if is_stringish(self.attributes["flow"]["project"]):
                      result["project"] = self.attributes["flow"][
                          "project"
                      ]
                  else:
                      raise MetaflowException(
                          "The *project* attribute of the *flow* is not a string"
                      )
              if "project_branch" in self.attributes["flow"]:
                  if is_stringish(
                      self.attributes["flow"]["project_branch"]
                  ):
                      result["branch"] = self.attributes["flow"][
                          "project_branch"
                      ]
                  else:
                      raise MetaflowException(
                          "The *project_branch* attribute of the *flow* is not a string"
                      )
              self.triggers.append(result)
          else:
              raise MetaflowException(
                  "Incorrect type for *flow* attribute in *@trigger_on_finish* "
                  " decorator. Supported type is string or Dict[str, str] - \n"
                  "@trigger_on_finish(flow='FooFlow') or "
                  "@trigger_on_finish(flow={'name':'FooFlow', 'project_branch': 'branch'})"
              )
      elif self.attributes["flows"]:
          # flows attribute supports the following formats -
          #     1. flows=['FooFlow', 'BarFlow']
          if isinstance(self.attributes["flows"], list):
              for flow in self.attributes["flows"]:
                  if is_stringish(flow):
                      self.triggers.append(
                          {
                              "fq_name": flow,
                          }
                      )
                  elif isinstance(flow, dict):
                      if "name" not in flow:
                          raise MetaflowException(
                              "One or more flows in the *flows* attribute for "
                              "*@trigger_on_finish* is missing the "
                              "*name* key."
                          )
                      flow_name = flow["name"]
  
                      if (
                          not is_stringish(flow_name)
                          or "." in flow_name
                      ):
                          raise MetaflowException(
                              "The *name* attribute '%s' is not a valid string"
                              % str(flow_name)
                          )
                      result = {"fq_name": flow_name}
                      if "project" in flow:
                          if is_stringish(flow["project"]):
                              result["project"] = flow["project"]
                          else:
                              raise MetaflowException(
                                  "The *project* attribute of the *flow* '%s' is not "
                                  "a string" % flow_name
                              )
                      if "project_branch" in flow:
                          if is_stringish(flow["project_branch"]):
                              result["branch"] = flow["project_branch"]
                          else:
                              raise MetaflowException(
                                  "The *project_branch* attribute of the *flow* %s "
                                  "is not a string" % flow_name
                              )
                      self.triggers.append(result)
                  else:
                      raise MetaflowException(
                          "One or more flows in *flows* attribute in "
                          "*@trigger_on_finish* decorator have an incorrect type. "
                          "Supported type is string or Dict[str, str]- \n"
                          "@trigger_on_finish(flows=['FooFlow', 'BarFlow']"
                      )
          else:
              raise MetaflowException(
                  "Incorrect type for *flows* attribute in *@trigger_on_finish* "
                  "decorator. Supported type is list - \n"
                  "@trigger_on_finish(flows=['FooFlow', 'BarFlow']"
              )
  
      if not self.triggers:
          raise MetaflowException(
              "No flow(s) specified in *@trigger_on_finish* decorator."
          )
  
      # Make triggers @project aware
      for trigger in self.triggers:
          if trigger["fq_name"].count(".") == 0:
              # fully qualified name is just the flow name
              trigger["flow"] = trigger["fq_name"]
          elif trigger["fq_name"].count(".") >= 2:
              # fully qualified name is of the format - project.branch.flow_name
              trigger["project"], tail = trigger["fq_name"].split(
                  ".", maxsplit=1
              )
              trigger["branch"], trigger["flow"] = tail.rsplit(
                  ".", maxsplit=1
              )
          else:
              raise MetaflowException(
                  "Incorrect format for *flow* in *@trigger_on_finish* "
                  "decorator. Specify either just the *flow_name* or a fully "
                  "qualified name like *project_name.branch_name.flow_name*."
              )
          # TODO: Also sanity check project and branch names
          if not re.match(r"^[A-Za-z0-9_]+$", trigger["flow"]):
              raise MetaflowException(
                  "Invalid flow name *%s* in *@trigger_on_finish* "
                  "decorator. Only alphanumeric characters and "
                  "underscores(_) are allowed." % trigger["flow"]
              )
  
      self.options = self.attributes["options"]
  
      # Handle scenario for local testing using --trigger.
      self._option_values = options
      if options["trigger"]:
          from metaflow import Run
          from metaflow.events import Trigger
  
          run_objs = []
          for run_pathspec in options["trigger"]:
              if len(run_pathspec.split("/")) != 2:
                  raise MetaflowException(
                      "Incorrect format for run pathspec for *--trigger*. "
                      "Supported format is flow_name/run_id."
                  )
              run_obj = Run(run_pathspec, _namespace_check=False)
              if not run_obj.successful:
                  raise MetaflowException(
                      "*--trigger* does not support runs that are not successful yet."
                  )
              run_objs.append(run_obj)
          current._update_env({"trigger": Trigger.from_runs(run_objs)})
  
file_path: metaflow/metaflow/plugins/events_decorator.py
imports: import json
  import time
  import re
  from metaflow import current
  from metaflow.decorators import FlowDecorator
  from metaflow.exception import MetaflowException
  from metaflow.util import is_stringish
label_desc: Use image node_id 1 for calling the TriggerOnFinishDecorator obj's underlying member method code with example usage: obj.flow_init(flow_name, graph, environment, flow_datastore, metadata, logger, echo, options) without return types