docs: def handle_finish(self, obj):
          """Handle an incoming ``Data`` finished processing request.
  
          :param obj: The Channels message object. Command object format:
  
              .. code-block:: none
  
                  {
                      'command': 'finish',
                      'data_id': [id of the :class:`~resolwe.flow.models.Data` object
                                 this command changes],
                      'process_rc': [exit status of the processing]
                      'spawn_processes': [optional; list of spawn dictionaries],
                      'exported_files_mapper': [if spawn_processes present]
                  }
          """
          data_id = obj[ExecutorProtocol.DATA_ID]
          logger.debug(
              __("Finishing Data with id {} (handle_finish).", data_id),
              extra={
                  'data_id': data_id,
                  'packet': obj
              }
          )
          spawning_failed = False
          with transaction.atomic():
              # Spawn any new jobs in the request.
              spawned = False
              if ExecutorProtocol.FINISH_SPAWN_PROCESSES in obj:
                  if is_testing():
                      # NOTE: This is a work-around for Django issue #10827
                      # (https://code.djangoproject.com/ticket/10827), same as in
                      # TestCaseHelpers._pre_setup(). Because the listener is running
                      # independently, it must clear the cache on its own.
                      ContentType.objects.clear_cache()
  
                  spawned = True
                  exported_files_mapper = obj[ExecutorProtocol.FINISH_EXPORTED_FILES]
                  logger.debug(
                      __("Spawning new Data objects for Data with id {} (handle_finish).", data_id),
                      extra={
                          'data_id': data_id
                      }
                  )
  
                  try:
                      # This transaction is needed because we're running
                      # asynchronously with respect to the main Django code
                      # here; the manager can get nudged from elsewhere.
                      with transaction.atomic():
                          parent_data = Data.objects.get(pk=data_id)
  
                          # Spawn processes.
                          for d in obj[ExecutorProtocol.FINISH_SPAWN_PROCESSES]:
                              d['contributor'] = parent_data.contributor
                              d['process'] = Process.objects.filter(slug=d['process']).latest()
                              d['tags'] = parent_data.tags
  
                              for field_schema, fields in iterate_fields(d.get('input', {}), d['process'].input_schema):
                                  type_ = field_schema['type']
                                  name = field_schema['name']
                                  value = fields[name]
  
                                  if type_ == 'basic:file:':
                                      fields[name] = self.hydrate_spawned_files(
                                          exported_files_mapper, value, data_id
                                      )
                                  elif type_ == 'list:basic:file:':
                                      fields[name] = [self.hydrate_spawned_files(exported_files_mapper, fn, data_id)
                                                      for fn in value]
  
                              with transaction.atomic():
                                  d = Data.objects.create(**d)
                                  DataDependency.objects.create(
                                      parent=parent_data,
                                      child=d,
                                      kind=DataDependency.KIND_SUBPROCESS,
                                  )
  
                                  # Copy permissions.
                                  copy_permissions(parent_data, d)
  
                                  # Entity is added to the collection only when it is
                                  # created - when it only contains 1 Data object.
                                  entities = Entity.objects.filter(data=d).annotate(num_data=Count('data')).filter(
                                      num_data=1)
  
                                  # Copy collections.
                                  for collection in parent_data.collection_set.all():
                                      collection.data.add(d)
  
                                      # Add entities to which data belongs to the collection.
                                      for entity in entities:
                                          entity.collections.add(collection)
  
                  except Exception:  # pylint: disable=broad-except
                      logger.error(
                          __(
                              "Error while preparing spawned Data objects of process '{}' (handle_finish):\n\n{}",
                              parent_data.process.slug,
                              traceback.format_exc()
                          ),
                          extra={
                              'data_id': data_id
                          }
                      )
                      spawning_failed = True
  
              # Data wrap up happens last, so that any triggered signals
              # already see the spawned children. What the children themselves
              # see is guaranteed by the transaction we're in.
              if ExecutorProtocol.FINISH_PROCESS_RC in obj:
                  process_rc = obj[ExecutorProtocol.FINISH_PROCESS_RC]
  
                  try:
                      d = Data.objects.get(pk=data_id)
                  except Data.DoesNotExist:
                      logger.warning(
                          "Data object does not exist (handle_finish).",
                          extra={
                              'data_id': data_id,
                          }
                      )
                      async_to_sync(self._send_reply)(obj, {ExecutorProtocol.RESULT: ExecutorProtocol.RESULT_ERROR})
                      return
  
                  changeset = {
                      'process_progress': 100,
                      'finished': now(),
                  }
  
                  if spawning_failed:
                      changeset['status'] = Data.STATUS_ERROR
                      changeset['process_error'] = ["Error while preparing spawned Data objects"]
  
                  elif process_rc == 0 and not d.status == Data.STATUS_ERROR:
                      changeset['status'] = Data.STATUS_DONE
  
                  else:
                      changeset['status'] = Data.STATUS_ERROR
                      changeset['process_rc'] = process_rc
  
                  obj[ExecutorProtocol.UPDATE_CHANGESET] = changeset
                  self.handle_update(obj, internal_call=True)
  
          if not getattr(settings, 'FLOW_MANAGER_KEEP_DATA', False):
              # Purge worker is not running in test runner, so we should skip triggering it.
              if not is_testing():
                  channel_layer = get_channel_layer()
                  try:
                      async_to_sync(channel_layer.send)(
                          CHANNEL_PURGE_WORKER,
                          {
                              'type': TYPE_PURGE_RUN,
                              'location_id': d.location.id,
                              'verbosity': self._verbosity,
                          }
                      )
                  except ChannelFull:
                      logger.warning(
                          "Cannot trigger purge because channel is full.",
                          extra={'data_id': data_id}
                      )
  
          # Notify the executor that we're done.
          async_to_sync(self._send_reply)(obj, {ExecutorProtocol.RESULT: ExecutorProtocol.RESULT_OK})
  
          # Now nudge the main manager to perform final cleanup. This is
          # needed even if there was no spawn baggage, since the manager
          # may need to know when executors have finished, to keep count
          # of them and manage synchronization.
          async_to_sync(consumer.send_event)({
              WorkerProtocol.COMMAND: WorkerProtocol.FINISH,
              WorkerProtocol.DATA_ID: data_id,
              WorkerProtocol.FINISH_SPAWNED: spawned,
              WorkerProtocol.FINISH_COMMUNICATE_EXTRA: {
                  'executor': getattr(settings, 'FLOW_EXECUTOR', {}).get('NAME', 'resolwe.flow.executors.local'),
              },
          })