Repository: spotify/luigi
File: luigi/task.py (Python)

Code before:
                        return False
                    paths = [task.output().path for task in large_chunk_of_tasks]
                    basenames = os.listdir(os.path.dirname(paths[0]))  # a single fs call
                    return all(os.path.basename(path) in basenames for path in paths)

                yield DynamicRequirements(large_chunk_of_tasks, custom_complete)

    .. py:attribute:: requirements

        The original, wrapped requirements.

    .. py:attribute:: flat_requirements

        Flattened view of the wrapped requirements (via :py:func:`flatten`). Read only.

    .. py:attribute:: paths

        Outputs of the requirements in the identical structure (via :py:func:`getpaths`). Read only.

    .. py:attribute:: custom_complete

       The optional, custom function performing the completeness check of the wrapped requirements.
    """

    def __init__(self, requirements, custom_complete=None):
        super(DynamicRequirements, self).__init__()

        # store attributes
        self.requirements = requirements
        self.custom_complete = custom_complete

        # cached flat requirements and paths
        self._flat_requirements = None
        self._paths = None

    @property
    def flat_requirements(self):
        if self._flat_requirements is None:
            self._flat_requirements = flatten(self.requirements)
        return self._flat_requirements

    @property
    def paths(self):
        if self._paths is None:
            self._paths = getpaths(self.requirements)
        return self._paths

    def complete(self, complete_fn=None):
        # default completeness check
        if complete_fn is None:
            def complete_fn(task):

Review comment (suggestion):
since Luigi only officially supports Python 3 now, this could be written `super().__init__()`. No need to change. Just pointing out.

Code after:
                        return False
                    paths = [task.output().path for task in large_chunk_of_tasks]
                    basenames = os.listdir(os.path.dirname(paths[0]))  # a single fs call
                    return all(os.path.basename(path) in basenames for path in paths)

                yield DynamicRequirements(large_chunk_of_tasks, custom_complete)

    .. py:attribute:: requirements

        The original, wrapped requirements.

    .. py:attribute:: flat_requirements

        Flattened view of the wrapped requirements (via :py:func:`flatten`). Read only.

    .. py:attribute:: paths

        Outputs of the requirements in the identical structure (via :py:func:`getpaths`). Read only.

    .. py:attribute:: custom_complete

       The optional, custom function performing the completeness check of the wrapped requirements.
    """

    def __init__(self, requirements, custom_complete=None):
        super().__init__()

        # store attributes
        self.requirements = requirements
        self.custom_complete = custom_complete

        # cached flat requirements and paths
        self._flat_requirements = None
        self._paths = None

    @property
    def flat_requirements(self):
        if self._flat_requirements is None:
            self._flat_requirements = flatten(self.requirements)
        return self._flat_requirements

    @property
    def paths(self):
        if self._paths is None:
            self._paths = getpaths(self.requirements)
        return self._paths

    def complete(self, complete_fn=None):
        # default completeness check
        if complete_fn is None:
            def complete_fn(task):