class_docstr: Specifies the event(s) that this flow depends on.
  
  ```
  @trigger(event='foo')
  ```
  or
  ```
  @trigger(events=['foo', 'bar'])
  ```
  
  Additionally, you can specify the parameter mappings
  to map event payload to Metaflow parameters for the flow.
  ```
  @trigger(event={'name':'foo', 'parameters':{'flow_param': 'event_field'}})
  ```
  or
  ```
  @trigger(events=[{'name':'foo', 'parameters':{'flow_param_1': 'event_field_1'},
                   {'name':'bar', 'parameters':{'flow_param_2': 'event_field_2'}])
  ```
  
  'parameters' can also be a list of strings and tuples like so:
  ```
  @trigger(event={'name':'foo', 'parameters':['common_name', ('flow_param', 'event_field')]})
  ```
  This is equivalent to:
  ```
  @trigger(event={'name':'foo', 'parameters':{'common_name': 'common_name', 'flow_param': 'event_field'}})
  ```
  
  Parameters
  ----------
  event : Union[str, dict], optional
      Event dependency for this flow.
  events : List[Union[str, dict]], optional
      Events dependency 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["event"], self.attributes["events"]),
              )
          )
          > 1
      ):
          raise MetaflowException(
              "Specify only one of *event* or *events* "
              "attributes in *@trigger* decorator."
          )
      elif self.attributes["event"]:
          # event attribute supports the following formats -
          #     1. event='table.prod_db.members'
          #     2. event={'name': 'table.prod_db.members',
          #               'parameters': {'alpha': 'member_weight'}}
          if is_stringish(self.attributes["event"]):
              self.triggers.append(
                  {"name": str(self.attributes["event"])}
              )
          elif isinstance(self.attributes["event"], dict):
              if "name" not in self.attributes["event"]:
                  raise MetaflowException(
                      "The *event* attribute for *@trigger* is missing the "
                      "*name* key."
                  )
              param_value = self.attributes["event"].get(
                  "parameters", {}
              )
              if isinstance(param_value, (list, tuple)):
                  new_param_value = {}
                  for mapping in param_value:
                      if is_stringish(mapping):
                          new_param_value[mapping] = mapping
                      elif (
                          isinstance(mapping, (list, tuple))
                          and len(mapping) == 2
                      ):
                          new_param_value[mapping[0]] = mapping[1]
                      else:
                          raise MetaflowException(
                              "The *parameters* attribute for event '%s' is invalid. "
                              "It should be a list/tuple of strings and lists/tuples "
                              "of size 2"
                              % self.attributes["event"]["name"]
                          )
                  self.attributes["event"][
                      "parameters"
                  ] = new_param_value
              self.triggers.append(self.attributes["event"])
          else:
              raise MetaflowException(
                  "Incorrect format for *event* attribute in *@trigger* decorator. "
                  "Supported formats are string and dictionary - \n"
                  "@trigger(event='foo') or @trigger(event={'name': 'foo', "
                  "'parameters': {'alpha': 'beta'}})"
              )
      elif self.attributes["events"]:
          # events attribute supports the following formats -
          #     1. events=[{'name': 'table.prod_db.members',
          #               'parameters': {'alpha': 'member_weight'}},
          #                {'name': 'table.prod_db.metadata',
          #               'parameters': {'beta': 'grade'}}]
          if isinstance(self.attributes["events"], list):
              for event in self.attributes["events"]:
                  if is_stringish(event):
                      self.triggers.append({"name": str(event)})
                  elif isinstance(event, dict):
                      if "name" not in event:
                          raise MetaflowException(
                              "One or more events in *events* attribute for "
                              "*@trigger* are missing the *name* key."
                          )
                      param_value = event.get("parameters", {})
                      if isinstance(param_value, (list, tuple)):
                          new_param_value = {}
                          for mapping in param_value:
                              if is_stringish(mapping):
                                  new_param_value[mapping] = mapping
                              elif (
                                  isinstance(mapping, (list, tuple))
                                  and len(mapping) == 2
                              ):
                                  new_param_value[mapping[0]] = mapping[
                                      1
                                  ]
                              else:
                                  raise MetaflowException(
                                      "The *parameters* attribute for event '%s' is "
                                      "invalid. It should be a list/tuple of strings "
                                      "and lists/tuples of size 2"
                                      % event["name"]
                                  )
                          event["parameters"] = new_param_value
                      self.triggers.append(event)
                  else:
                      raise MetaflowException(
                          "One or more events in *events* attribute in *@trigger* "
                          "decorator have an incorrect format. Supported format "
                          "is dictionary - \n"
                          "@trigger(events=[{'name': 'foo', 'parameters': {'alpha': "
                          "'beta'}}, {'name': 'bar', 'parameters': "
                          "{'gamma': 'kappa'}}])"
                      )
          else:
              raise MetaflowException(
                  "Incorrect format for *events* attribute in *@trigger* decorator. "
                  "Supported format is list - \n"
                  "@trigger(events=[{'name': 'foo', 'parameters': {'alpha': "
                  "'beta'}}, {'name': 'bar', 'parameters': "
                  "{'gamma': 'kappa'}}])"
              )
  
      if not self.triggers:
          raise MetaflowException(
              "No event(s) specified in *@trigger* decorator."
          )
  
      # same event shouldn't occur more than once
      names = [x["name"] for x in self.triggers]
      if len(names) != len(set(names)):
          raise MetaflowException(
              "Duplicate event names defined in *@trigger* decorator."
          )
  
      self.options = self.attributes["options"]
  
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 TriggerDecorator 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