@classmethod
  def StartFlow(cls, args=None, runner_args=None,  # pylint: disable=g-bad-name
                parent_flow=None, _store=None, sync=True, **kwargs):
    """The main factory function for Creating and executing a new flow.

    Args:

      args: An arg protocol buffer which is an instance of the required flow's
        args_type class attribute.

      runner_args: an instance of FlowRunnerArgs() protocol buffer which is used
        to initialize the runner for this flow.

      parent_flow: A parent flow or None if this is a top level flow.
      _store: The data store to use for running this flow (only used for
              testing).

      sync: If True, the Start method of this flow will be called
         inline. Otherwise we schedule the starting of this flow on another
         worker.

      **kwargs: If args or runner_args are not specified, we construct these
        protobufs from these keywords.

    Returns:
      the session id of the flow.

    Raises:
      RuntimeError: Unknown or invalid parameters were provided.
    """

    # Build the runner args from the keywords.
    if runner_args is None:
      runner_args = flow_runner.FlowRunnerArgs()

    cls.FilterArgsFromSemanticProtobuf(runner_args, kwargs)

    # When asked to run a flow in the future this implied it will run
    # asynchronously.
    if runner_args.start_time:
      sync = False

    # Is the required flow a known flow?
    if runner_args.flow_name not in GRRFlow.classes:
      stats.STATS.IncrementCounter("grr_flow_invalid_flow_count")
      raise RuntimeError("Unable to locate flow %s" % runner_args.flow_name)

    # If no token is specified, use the default token.
    if not runner_args.HasField("token"):
      if data_store.default_token is None:
        raise access_control.UnauthorizedAccess("A token must be specified.")

      runner_args.token = data_store.default_token.Copy()

    # Make sure we are allowed to run this flow. If not, we raise here. We
    # respect SUID (supervisor) if it is already set. SUID cannot be set by the
    # user since it isn't part of the ACLToken proto.
    data_store.DB.security_manager.CheckIfCanStartFlow(
        runner_args.token, runner_args.flow_name,
        with_client_id=runner_args.client_id)

    flow_cls = GRRFlow.GetPlugin(runner_args.flow_name)
    # If client id was specified and flow doesn't have exemption from ACL
    # checking policy, then check that the user has access to the client
    # where the flow is going to run.
    if flow_cls.ACL_ENFORCED and runner_args.client_id:
      data_store.DB.security_manager.CheckClientAccess(runner_args.token,
                                                       runner_args.client_id)

    # For the flow itself we use a supervisor token.
    token = runner_args.token.SetUID()

    # Extend the expiry time of this token indefinitely. Python on Windows only
    # supports dates up to the year 3000, this number corresponds to July, 2997.
    token.expiry = 32427003069 * rdfvalue.RDFDatetime.converter

    # We create an anonymous AFF4 object first, The runner will then generate
    # the appropriate URN.
    flow_obj = aff4.FACTORY.Create(None, runner_args.flow_name,
                                   token=token)

    # Now parse the flow args into the new object from the keywords.
    if args is None:
      args = flow_obj.args_type()

    cls.FilterArgsFromSemanticProtobuf(args, kwargs)

    # Check that the flow args are valid.
    args.Validate()

    # Store the flow args in the state.
    flow_obj.state.Register("args", args)

    # A convenience attribute to allow flows to access their args directly.
    flow_obj.args = args

    # At this point we should exhaust all the keyword args. If any are left
    # over, we do not know what to do with them so raise.
    if kwargs:
      raise type_info.UnknownArg(
          "Unknown parameters to StartFlow: %s" % kwargs)

    # Create a flow runner to run this flow with.
    if parent_flow:
      parent_runner = parent_flow.runner
    else:
      parent_runner = None

    runner = flow_obj.CreateRunner(parent_runner=parent_runner,
                                   runner_args=runner_args,
                                   _store=_store or data_store.DB)

    logging.info(u"Scheduling %s(%s) on %s", flow_obj.urn,
                 runner_args.flow_name, runner_args.client_id)

    if sync:
      # Just run the first state inline. NOTE: Running synchronously means
      # that this runs on the thread that starts the flow. The advantage is
      # that that Start method can raise any errors immediately.
      flow_obj.Start()
    else:
      # Running Asynchronously: Schedule the start method on another worker.
      runner.CallState(next_state="Start", start_time=runner_args.start_time)

    # The flow does not need to actually remain running.
    if not runner.OutstandingRequests():
      flow_obj.Terminate()

    flow_obj.Close()

    # Publish an audit event, only for top level flows.
    if parent_flow is None:
      Events.PublishEvent("Audit",
                          AuditEvent(user=token.username,
                                     action="RUN_FLOW",
                                     flow_name=runner_args.flow_name,
                                     urn=flow_obj.urn,
                                     client=runner_args.client_id),
                          token=token)

    return flow_obj.urn
dataset/ETHPy150Open google/grr/grr/lib/flow.py/GRRFlow.StartFlow/original