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.

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