def handle(self):
    """Handle request.

    This method has to be careful to pass the same ShardState instance to
    its subroutines calls if the calls mutate or read from ShardState.
    Note especially that Context instance caches and updates the ShardState
    instance.

    Returns:
      Set HTTP status code and always returns None.
    """
    # Reconstruct basic states.
    self._start_time = self._time()
    shard_id = self.request.headers[util._MR_SHARD_ID_TASK_HEADER]
    mr_id = self.request.headers[util._MR_ID_TASK_HEADER]
    spec = model.MapreduceSpec._get_mapreduce_spec(mr_id)
    shard_state, control = db.get([
        model.ShardState.get_key_by_shard_id(shard_id),
        model.MapreduceControl.get_key_by_job_id(mr_id),
    ])

    # Set context before any IO code is called.
    ctx = context.Context(spec, shard_state,
                          task_retry_count=self.task_retry_count())
    context.Context._set(ctx)

    # Unmarshall input reader, output writer, and other transient states.
    tstate = model.TransientShardState.from_request(self.request)

    # Try acquire a lease on the shard.
    if shard_state:
      is_this_a_retry = shard_state.acquired_once
    task_directive = self._try_acquire_lease(shard_state, tstate)
    if task_directive in (self._TASK_DIRECTIVE.RETRY_TASK,
                          self._TASK_DIRECTIVE.DROP_TASK):
      return self.__return(shard_state, tstate, task_directive)
    assert task_directive == self._TASK_DIRECTIVE.PROCEED_TASK

    # Abort shard if received signal.
    if control and control.command == model.MapreduceControl.ABORT:
      task_directive = self._TASK_DIRECTIVE.ABORT_SHARD
      return self.__return(shard_state, tstate, task_directive)

    # Retry shard if user disabled slice retry.
    if (is_this_a_retry and
        1 <= parameters.config.TASK_MAX_DATA_PROCESSING_ATTEMPTS):
      task_directive = self._TASK_DIRECTIVE.RETRY_SHARD
      return self.__return(shard_state, tstate, task_directive)

    # TODO(user): Find a better way to set these per thread configs.
    # E.g. what if user change it?
    util._set_ndb_cache_policy()

    job_config = map_job.JobConfig._to_map_job_config(
        spec,
        os.environ.get("HTTP_X_APPENGINE_QUEUENAME"))
    job_context = map_job_context.JobContext(job_config)
    self.shard_context = map_job_context.ShardContext(job_context, shard_state)
    self.slice_context = map_job_context.SliceContext(self.shard_context,
                                                      shard_state,
                                                      tstate)
    try:
      slice_id = tstate.slice_id
      self._lc_start_slice(tstate, slice_id)

      if shard_state.is_input_finished():
        self._lc_end_slice(tstate, slice_id, last_slice=True)
        # Finalize the stream and set status if there's no more input.
        if (tstate.output_writer and
            isinstance(tstate.output_writer, output_writers.OutputWriter)):
          # It's possible that finalization is successful but
          # saving state failed. In this case this shard will retry upon
          # finalization error.
          # TODO(user): make finalize method idempotent!
          tstate.output_writer.finalize(ctx, shard_state)
        shard_state.set_for_success()
        return self.__return(shard_state, tstate, task_directive)

      if is_this_a_retry:
        task_directive = self._attempt_slice_recovery(shard_state, tstate)
        if task_directive != self._TASK_DIRECTIVE.PROCEED_TASK:
          return self.__return(shard_state, tstate, task_directive)

      last_slice = self._process_inputs(
          tstate.input_reader, shard_state, tstate, ctx)

      self._lc_end_slice(tstate, slice_id)

      ctx.flush()

      if last_slice:
        # We're done processing data but we still need to finalize the output
        # stream. We save this condition in datastore and force a new slice.
        # That way if finalize fails no input data will be retried.
        shard_state.set_input_finished()
    # pylint: disable=broad-except
    except Exception, e:
      logging.warning("Shard %s got error.", shard_state.shard_id)
      logging.error(traceback.format_exc())

      # Fail fast.
      if type(e) is errors.FailJobError:
        logging.error("Got FailJobError.")
        task_directive = self._TASK_DIRECTIVE.FAIL_TASK
      else:
        task_directive = self._TASK_DIRECTIVE.RETRY_SLICE

    self.__return(shard_state, tstate, task_directive)
dataset/ETHPy150Open GoogleCloudPlatform/appengine-mapreduce/python/src/mapreduce/handlers.py/MapperWorkerCallbackHandler.handle/swapped operands of `<=`