docs: def handle_update(self, obj, internal_call=False):
          """Handle an incoming ``Data`` object update request.
  
          :param obj: The Channels message object. Command object format:
  
              .. code-block:: none
  
                  {
                      'command': 'update',
                      'data_id': [id of the :class:`~resolwe.flow.models.Data`
                                 object this command changes],
                      'changeset': {
                          [keys to be changed]
                      }
                  }
  
          :param internal_call: If ``True``, this is an internal delegate
              call, so a reply to the executor won't be sent.
          """
          data_id = obj[ExecutorProtocol.DATA_ID]
          changeset = obj[ExecutorProtocol.UPDATE_CHANGESET]
          if not internal_call:
              logger.debug(
                  __("Handling update for Data with id {} (handle_update).", data_id),
                  extra={
                      'data_id': data_id,
                      'packet': obj
                  }
              )
          try:
              d = Data.objects.get(pk=data_id)
          except Data.DoesNotExist:
              logger.warning(
                  "Data object does not exist (handle_update).",
                  extra={
                      'data_id': data_id,
                  }
              )
  
              if not internal_call:
                  async_to_sync(self._send_reply)(obj, {ExecutorProtocol.RESULT: ExecutorProtocol.RESULT_ERROR})
  
              async_to_sync(consumer.send_event)({
                  WorkerProtocol.COMMAND: WorkerProtocol.ABORT,
                  WorkerProtocol.DATA_ID: obj[ExecutorProtocol.DATA_ID],
                  WorkerProtocol.FINISH_COMMUNICATE_EXTRA: {
                      'executor': getattr(settings, 'FLOW_EXECUTOR', {}).get('NAME', 'resolwe.flow.executors.local'),
                  },
              })
  
              return
  
          if changeset.get('status', None) == Data.STATUS_ERROR:
              logger.error(
                  __("Error occured while running process '{}' (handle_update).", d.process.slug),
                  extra={
                      'data_id': data_id,
                      'api_url': '{}{}'.format(
                          getattr(settings, 'RESOLWE_HOST_URL', ''),
                          reverse('resolwe-api:data-detail', kwargs={'pk': data_id})
                      ),
                  }
              )
  
          if d.status == Data.STATUS_ERROR:
              changeset['status'] = Data.STATUS_ERROR
  
          if not d.started:
              changeset['started'] = now()
          changeset['modified'] = now()
  
          for key, val in changeset.items():
              if key in ['process_error', 'process_warning', 'process_info']:
                  # Trim process_* fields to not exceed max length of the database field.
                  for i, entry in enumerate(val):
                      max_length = Data._meta.get_field(key).base_field.max_length  # pylint: disable=protected-access
                      if len(entry) > max_length:
                          val[i] = entry[:max_length - 3] + '...'
  
                  getattr(d, key).extend(val)
  
              elif key != 'output':
                  setattr(d, key, val)
  
          if 'output' in changeset:
              if not isinstance(d.output, dict):
                  d.output = {}
              for key, val in changeset['output'].items():
                  dict_dot(d.output, key, val)
  
          try:
              d.save(update_fields=list(changeset.keys()))
          except ValidationError as exc:
              logger.error(
                  __(
                      "Validation error when saving Data object of process '{}' (handle_update):\n\n{}",
                      d.process.slug,
                      traceback.format_exc()
                  ),
                  extra={
                      'data_id': data_id
                  }
              )
  
              d.refresh_from_db()
  
              d.process_error.append(exc.message)
              d.status = Data.STATUS_ERROR
  
              try:
                  d.save(update_fields=['process_error', 'status'])
              except Exception:  # pylint: disable=broad-except
                  pass
          except Exception:  # pylint: disable=broad-except
              logger.error(
                  __(
                      "Error when saving Data object of process '{}' (handle_update):\n\n{}",
                      d.process.slug,
                      traceback.format_exc()
                  ),
                  extra={
                      'data_id': data_id
                  }
              )
  
          if not internal_call:
              async_to_sync(self._send_reply)(obj, {ExecutorProtocol.RESULT: ExecutorProtocol.RESULT_OK})