code: def begin_abort(self, root_pipeline_key, abort_message):
      
      def txn():
        pipeline_record = db.get(root_pipeline_key)
        if pipeline_record is None:
          logging.warning(
              'Tried to abort root pipeline ID "%s" but it does not exist.',
              root_pipeline_key.name())
          raise db.Rollback()
        if pipeline_record.status == _PipelineRecord.ABORTED:
          logging.warning(
              'Tried to abort root pipeline ID "%s"; already in state: %s',
              root_pipeline_key.name(), pipeline_record.status)
          raise db.Rollback()
        if pipeline_record.abort_requested:
          logging.warning(
              'Tried to abort root pipeline ID "%s"; abort signal already sent.',
              root_pipeline_key.name())
          raise db.Rollback()
  
        pipeline_record.abort_requested = True
        pipeline_record.abort_message = abort_message
        pipeline_record.put()
  
        task = taskqueue.Task(
            url=self.fanout_abort_handler_path,
            params=dict(root_pipeline_key=root_pipeline_key))
        task.add(queue_name=self.queue_name, transactional=True)
        return True
  
      return db.run_in_transaction(txn)
docs: Kicks off the abort process for a root pipeline and all its children.
  
      Args:
        root_pipeline_key: db.Key of the root pipeline to abort.
        abort_message: Message explaining why the abort happened, only saved
            into the root pipeline.
  
      Returns:
        True if the abort signal was sent successfully; False otherwise.