docs: def check_completion(task, mark_incomplete=False, clear=False, return_stats=False):
      """Recursively check if a task and all its requirements are complete
  
      Args:
          task (derived from luigi.Task): Task to check completion for; check everything 'downstream'
              from that task.
  
          mark_incomplete (bool): If ``True`` set any task as incomplete for which a requirement
              is found to be incomplete (checked recursively).
              This works only for tasks derived from :class:`ORMTask`.
  
          clear (bool): If ``True``, call the :func:`clear()` method of any task for which a requirement
              is found to be incomplete (checked recursively). This implies ``mark_incomplete = True``.
              This works only for tasks derived from :class:`ORMTask`.
  
          return_stats (bool): If ``True``, return task checking statistics in addition to completion status
  
      Returns:
          bool: ``True`` if the task, all its requirements and (recursively) all their requirements
              are complete, ``False`` otherwise.
      """
      # run recursive task checking, get stats
      to_clear = dict()
      is_complete, stats = _check_completion(task,
                                             mark_incomplete=mark_incomplete,
                                             clear=clear,
                                             stats={},
                                             visited=dict(),
                                             to_clear=to_clear)
  
      # task clearing needs to happen top-down: because of foreign key constraints, a task can
      # only be cleared once all tasks that require it have been cleared
      while to_clear:
          # find all tasks that we can currently clear - tasks not required by other tasks;
          # iterate over list of keys to be able to modify dict while iterating
          found_clearable_task = False
          for task_id in list(to_clear.keys()):
              v = to_clear[task_id]
              if not v['required_by']:
                  # this is a task that can be cleared - no other task requires it
                  found_clearable_task = True
  
                  task = v['task']
                  if isinstance(task, ORMTask):
                      task.mark_incomplete()
                      task.clear()
                      _increment_stats(stats, 'Cleared')
                      config.logger.info("Cleared task: " + task_id)
                  else:
                      config.logger.info('Cannot clear task, not an ORMTask: ' + task_id)
  
                  # remove the task from the list of tasks that need clearing, remove references
                  # in the required_by lists of all other tasks; this is not an efficient implementation,
                  # O(n^2), could be made O(n) using lookup tables of the task graph
                  del to_clear[task_id]
                  for w in to_clear.values():
                      w['required_by'].discard(task_id)
  
          if not found_clearable_task:
              raise RuntimeError("Error in recursive task clearing, no clearable task found")
  
      config.logger.info("Task completion checking, summary:\n" + str(stats))
  
      if return_stats:
          return is_complete, stats
      else:
          return is_complete