class Backend:
    """Base class for backends."""

    option_defaults = {}
    log_manager = FileLogManager()

    @classmethod
    def list(cls):
        """Return the names of all registered backends."""
        return set(_load_backends().keys())

    @classmethod
    def from_name(cls, name):
        """Return backend class for the backend given by `name`.

        Returns the backend class registered with `name`. Note that the *class*
        is returned, not the instance, since not all uses requires
        initialization of the backend (e.g. accessing the backends' log
        manager), and initialization of the backend may be expensive.

        :arg str name: Path to a workflow file, optionally specifying a
            workflow object in that file.
        """
        return _load_backends()[name]

    @classmethod
    def from_config(cls, config):
        """Return backend class for the backend specified by `config`.

        See :func:`Backend.from_name` for further information."""
        return cls.from_name(config["backend"])

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def status(self, target):
        """Return the status of `target`.

        :param gwf.Target target: The target to return the status of.
        :return gwf.backends.Status: Status of `target`.
        """

    def submit_full(self, target, dependencies):
        """Prepare and submit `target` with `dependencies`.

        Will prepare the target for submission by injecting option defaults
        from the backend, check for unsupported options, and removing options
        with a `None` value.

        This is the primary way to submit a target. Do not call
        :func:`submit` directly, unless you want to manually deal with with
        injection of option defaults.
        """
        new_options = dict(self.option_defaults)
        new_options.update(target.options)

        for option_name, option_value in list(new_options.items()):
            if option_name not in self.option_defaults.keys():
                logger.warning(
                    "Option '%s' used in '%s' is not supported by backend. Ignored.",
                    option_name,
                    target.name,
                )
                del new_options[option_name]
            elif option_value is None:
                del new_options[option_name]
        target.options = new_options

        self.submit(target, dependencies)

    def submit(self, target, dependencies):
        """Submit `target` with `dependencies`.

        This method must submit the `target` and return immediately. That is,
        the method must not block while waiting for the target to complete.

        :param gwf.Target target:
            The target to submit.
        :param dependencies:
            An iterable of :class:`gwf.Target` objects that `target` depends on
            and that have already been submitted to the backend.
        """

    def cancel(self, target):
        """Cancel `target`.

        :param gwf.Target target:
            The target to cancel.
        :raises gwf.exception.TargetError:
            If the target does not exist in the workflow.
        """

    @classmethod
    def logs(cls, target, stderr=False):
        """Return log files for a target.

        If the backend cannot return logs a
        :class:`~gwf.exceptions.NoLogFoundError` is raised.

        By default standard output (stdout) is returned. If `stderr=True`
        standard error will be returned instead.

        :param gwf.Target target:
            Target to return logs for.
        :param bool stderr:
            default: False. If true, return standard error.
        :return:
            A file-like object. The user is responsible for closing the
            returned file(s) after use.
        :raises gwf.exceptions.NoLogFoundError:
            if the backend could not find a log for the given target.
        """
        if stderr:
            return cls.log_manager.open_stderr(target)
        return cls.log_manager.open_stdout(target)

    def close(self):
        """Close the backend.

        Called when the backend is no longer needed and should close all
        resources (open files, connections) used by the backend.
        """ class Graph:
    """Represents a dependency graph for a set of targets.

    The graph represents the targets present in a workflow, but also their
    dependencies and the files they provide.

    During construction of the graph the dependencies between targets are
    determined by looking at target inputs and outputs. If a target specifies a
    file as input, the file must either be provided by another target or
    already exist on disk. In case that the file is provided by another target,
    a dependency to that target will be added:

    :ivar dict dependencies:
        A dictionary mapping a target to a set of its dependencies.

    If the file is not provided by another target, the file is *unresolved*:

    :ivar set unresolved:
        A set containing file paths of all unresolved files.

    If the graph is constructed successfully, the following instance variables
    will be available:

    :ivar dict targets:
        A dictionary mapping target names to instances of :class:`gwf.Target`.
    :ivar dict provides:
        A dictionary mapping a file path to the target that provides that path.
    :ivar dict dependents:
        A dictionary mapping a target to a set of all targets which depend on
        the target.

    The graph can be manipulated in arbitrary, diabolic ways after it has been
    constructed. Checks are only performed at construction-time, thus
    introducing e.g. a circular dependency by manipulating *dependencies* will
    not raise an exception.

    :raises gwf.exceptions.WorkflowError:
        Raised if the workflow contains a circular dependency.
    """

    def __init__(self, targets, provides, dependencies, dependents, unresolved):
        self.targets = targets
        self.provides = provides
        self.dependencies = dependencies
        self.dependents = dependents
        self.unresolved = unresolved

        self._check_for_circular_dependencies()

    @classmethod
    def from_targets(cls, targets):
        """Construct a dependency graph from a set of targets.

        When a graph is initialized it computes all dependency relations
        between targets, ensuring that the graph is semantically sane.
        Therefore, construction of the graph is an expensive operation which
        may raise a number of exceptions:

        :raises gwf.exceptions.FileProvidedByMultipleTargetsError:
            Raised if the same file is provided by multiple targets.

        Since this method initializes the graph, it may also raise:

        :raises gwf.exceptions.WorkflowError:
            Raised if the workflow contains a circular dependency.
        """
        provides = {}
        unresolved = set()
        dependencies = defaultdict(set)
        dependents = defaultdict(set)

        logger.debug("Building dependency graph from %d targets", len(targets))

        if isinstance(targets, dict):
            targets = targets.values()

        with timer("Built dependency graph in %.3fms", logger=logger):
            for target in targets:
                for path in target.flattened_outputs():
                    if path in provides:
                        msg = 'File "{}" provided by targets "{}" and "{}".'.format(
                            path, provides[path].name, target
                        )
                        raise WorkflowError(msg)
                    provides[path] = target

        for target in targets:
            for path in target.flattened_inputs():
                if path in provides:
                    dependencies[target].add(provides[path])
                else:
                    unresolved.add(path)

        for target, deps in dependencies.items():
            for dep in deps:
                dependents[dep].add(target)

        return cls(
            targets={target.name: target for target in targets},
            provides=provides,
            dependencies=dependencies,
            dependents=dependents,
            unresolved=unresolved,
        )

    @timer("Checked for circular dependencies in %.3fms", logger=logger)
    def _check_for_circular_dependencies(self):
        """Check for circular dependencies in the graph.

        Raises :class:`WorkflowError` if a circular dependency is found.
        """
        logger.debug("Checking for circular dependencies")

        fresh, started, done = 0, 1, 2

        nodes = self.targets.values()
        state = dict((n, fresh) for n in nodes)

        def visitor(node):
            state[node] = started
            for dep in self.dependencies[node]:
                if state[dep] == started:
                    raise WorkflowError("Target {} depends on itself.".format(node))
                elif state[dep] == fresh:
                    visitor(dep)
            state[node] = done

        for node in nodes:
            if state[node] == fresh:
                visitor(node)

    def endpoints(self):
        """Return a set of all targets that are not depended on by other targets."""
        return set(self.targets.values()) - set(self.dependents.keys())

    @cache
    def dfs(self, root):
        """Return the depth-first traversal path through a graph from `root`."""
        visited = set()
        path = []

        def dfs_inner(node):
            if node in visited:
                return

            visited.add(node)
            for dep in self.dependencies[node]:
                dfs_inner(dep)
            path.append(node)

        dfs_inner(root)
        return path

    def subset(self, endpoints):
        """Subset a graph given an iterable of endpoints.

        This will return a new, potentially disjoint graph with the given
        endpoints and all of their dependencies.
        """
        targets = set()
        for target in endpoints:
            for dep in self.dfs(target):
                targets.add(dep)
        return Graph.from_targets(targets)

    def __iter__(self):
        return iter(self.targets.values())

    def __len__(self):
        return len(self.targets)

    def __getitem__(self, target_name):
        return self.targets[target_name]

    def __contains__(self, target_name):
        return target_name in self.targets class TargetStatus(Enum):
    """Status of a target as computed by the Scheduler."""

    SHOULDRUN = 0  #: The target should run.
    SUBMITTED = 1  #: The target has been submitted, but is not currently running.
    RUNNING = 2  #: The target is currently running.
    COMPLETED = 3  #: The target has completed and should not run. def get_status(target, scheduled, backend):
    """Return the status of a target.

    Returns the status of a target where it is taken into account whether
    the target should run or not.

    :param Target target:
        The target to return status for.
    """
    status = backend.status(target)
    if status == Status.UNKNOWN:
        if target in scheduled:
            return TargetStatus.SHOULDRUN
        return TargetStatus.COMPLETED
    return TargetStatus[status.name] def schedule(self, targets, graph):
    """Schedule multiple targets and their dependencies.

    :param list targets:
        A list of targets to be scheduled.
    """

    logger.debug("Scheduling %d target(s)", len(targets))
    with timer("Scheduled targets in %.3fms", logger=logger):
        for target in targets:
            self._schedule_dependencies(target, graph)

    return self._scheduled, self._reasons class EndpointFilter(ApplyMixin):
    def __init__(self, endpoints, mode="include"):
        self.endpoints = endpoints
        self.mode = mode

    def predicate(self, target):
        if self.mode == "exclude":
            return target not in self.endpoints
        elif self.mode == "include":
            return target in self.endpoints
        else:
            raise ValueError('Argument mode must be either "include" or "exclude".') class NameFilter:
    def __init__(self, patterns):
        self.patterns = patterns

    def apply(self, targets):
        target_name_map = {target.name: target for target in targets}
        return {
            target_name_map[name]
            for pattern in self.patterns
            for name in fnmatch.filter(target_name_map.keys(), pattern)
        } class StatusFilter(ApplyMixin):
    def __init__(self, status_provider, status):
        self.status_provider = status_provider
        if not hasattr(status, "__iter__"):
            status = [status]
        self.status = status

    def predicate(self, target):
        return self.status_provider(target) in self.status def filter_generic(targets, filters):
    """Filter targets given a list of filters.

    Return all targets from `targets` passing all `filters`. For example:

    .. code-block:: python

        matched_targets = filter_generic(
            targets=graph.targets.values(),
            filters=[
                NameFilter(patterns=['Foo*'],
                StatusFilter(scheduler=scheduler, status='running'),
            ]
        )

    returns a generator yielding all targets with a name matching ``Foo*``
    which are currently running.

    :arg targets: A list of targets to be filtered.

    :arg filters: A list of :class:`Filter` instances.
    """
    return CompositeFilter(filters).apply(targets) class Workflow(object):
    """Represents a workflow.

    This is the most central user-facing abstraction in *gwf*.

    A workflow consists of a collection of targets and has methods for adding
    targets to the workflow in two different ways. A workflow can be
    initialized with the following arguments:

    :ivar str name: initial value: None
        The name is used for namespacing when including workflows. See
        :func:`~include` for more details on namespacing.
    :ivar str working_dir:
        The directory containing the file where the workflow was initialized.
        All file paths used in targets added to this workflow are relative to
        the working directory.
    :ivar dict defaults:
        A dictionary with defaults for target options.

    By default, *working_dir* is set to the directory of the workflow file
    which initialized the workflow. However, advanced users may wish to set it
    manually. Targets added to the workflow will inherit the workflow working
    directory.

    The *defaults* argument is a dictionary of option defaults for targets and
    overrides defaults provided by the backend. Targets can override the
    defaults individually. For example::

        gwf = Workflow(defaults={
            'cores': 12,
            'memory': '16g',
        })

        gwf.target('Foo', inputs=[], outputs=[]) << \"\"\"echo hello\"\"\"
        gwf.target('Bar', inputs=[], outputs=[], cores=2) << \"\"\"echo world\"\"\"

    In this case `Foo` and `Bar` inherit the `cores` and `memory` options set
    in `defaults`, but `Bar` overrides the `cores` option.

    See :func:`~include` for a description of the use of the `name` argument.
    """

    def __init__(self, name=None, working_dir=None, defaults=None):
        self.name = name
        if self.name is not None and not is_valid_name(self.name):
            raise NameError(
                'Workflow defined with invalid name: "{}".'.format(self.name)
            )

        self.targets = {}
        self.defaults = defaults or {}

        self.working_dir = working_dir
        if self.working_dir is None:
            # Get the frame object of whatever called the Workflow.__init__
            # and extract the path of the file which is was defined in. Then
            # normalize the path and get the directory of the file.
            filename = inspect.getfile(sys._getframe(1))
            self.working_dir = os.path.dirname(os.path.realpath(filename))

    @classmethod
    def from_path(cls, path):
        """Return workflow object for the workflow given by `path`.

        Returns a :class:`~gwf.Workflow` object containing the workflow object
        of the workflow given by `path`.

        :arg str path: Path to a workflow file, optionally specifying a
            workflow object in that file.
        """
        basedir, filename, obj = parse_path(path)
        return load_workflow(basedir, filename, obj)

    @classmethod
    def from_config(cls, config):
        """Return workflow object for the workflow specified by `config`.

        See :func:`Workflow.from_path` for further information.
        """
        return cls.from_path(config["file"])

    def _add_target(self, target, namespace=None):
        if namespace is not None:
            target.name = target.qualname(namespace)
        if target.name in self.targets:
            raise WorkflowError(
                'Target "{}" already exists in workflow.'.format(target.name)
            )
        self.targets[target.name] = target

    def target(self, name, inputs, outputs, **options):
        """Create a target and add it to the :class:`gwf.Workflow`.

        This is syntactic sugar for creating a new :class:`~gwf.Target` and
        adding it to the workflow. The target is also returned from the method
        so that the user can directly manipulate it, if necessary. For example,
        this allows assigning a spec to a target directly after defining it::

            workflow = Workflow()
            workflow.target('NewTarget', inputs=['test.txt', 'out.txt']) <<< '''
            cat test.txt > out.txt
            echo hello world >> out.txt
            '''

        This will create a new target named `NewTarget`, add it to the workflow
        and assign a spec to the target.

        :param str name: Name of the target.
        :param iterable inputs: List of files that this target depends on.
        :param iterable outputs: List of files that this target produces.

        Any further keyword arguments are passed to the backend.
        """
        new_target = Target(
            name=name,
            inputs=inputs,
            outputs=outputs,
            options=options,
            working_dir=self.working_dir,
        )
        new_target.inherit_options(self.defaults)

        self._add_target(new_target)
        return new_target

    def target_from_template(self, name, template, **options):
        """Create a target from a template and add it to the :class:`gwf.Workflow`.

        This is syntactic sugar for creating a new :class:`~gwf.Target` and
        adding it to the workflow. The target is also returned from the method
        so that the user can directly manipulate it, if necessary.

        .. code-block:: python

            workflow = Workflow()
            workflow.target_from_template('NewTarget', my_template())

        This will create a new target named `NewTarget`, configure it based
        on the specification in the template `my_template`, and
        add it to the workflow.

        :param str name:
            Name of the target.
        :param AnonymousTarget template:
            The anonymous target which describes the template.

        Any further keyword arguments are passed to the backend and will
        override any options provided by the template.
        """
        if isinstance(template, AnonymousTarget):
            new_target = Target(
                name=name,
                inputs=template.inputs,
                outputs=template.outputs,
                options=options,
                working_dir=template.working_dir or self.working_dir,
                spec=template.spec,
            )

            new_target.inherit_options(template.options)
        elif isinstance(template, tuple):
            warnings.warn(
                (
                    "Creating a target from a tuple template is deprecated, "
                    "and will be removed in gwf 2.0. Make your template "
                    "function return an AnonymousTarget instead."
                ),
                DeprecationWarning,
            )

            try:
                inputs, outputs, template_options, spec = template
            except ValueError:
                raise TypeError(
                    "Target `{}` received an invalid template.".format(name)
                )

            new_target = Target(
                name=name,
                inputs=inputs,
                outputs=outputs,
                options=options,
                working_dir=self.working_dir,
                spec=spec,
            )

            new_target.inherit_options(template_options)
        else:
            raise TypeError("Target `{}` received an invalid template.".format(name))

        new_target.inherit_options(self.defaults)
        self._add_target(new_target)
        return new_target

    def map(self, template_func, inputs, extra=None, name=None, **kwargs):
        """Add targets to the workflow given a template and a list of inputs.

        This method accepts a template function and an iterable of inputs. For
        each item in `inputs` it produces a target using the template function
        and adds the target to this workflow.

        For example, given this template:

        .. code-block::

            def copy_file(from_file):
                inputs = {'from_file': from_file}
                outputs = {'to_file': to_file + '.copy'}
                options = {}
                spec = f"cp {inputs[from_file]} {outputs[to_file]}"
                return AnonymousTarget(
                    inputs=inputs,
                    outputs=outputs,
                    options=options,
                    spec=spec
                )

        and this list of files:

        .. code-block::

            files = ['file1', 'file2', 'file3']

        we can generate targets to copy all three files:

        .. code-block::

            gwf = Workflow()
            res = gwf.map(copy_file, files)

        The :func:`map` method returns a :class:`TargetList` which contains the
        generated targets.

        :param template_func:
            A function or callable class instance that returns an
            :class:`AnonymousTarget`. Essentially a *template function*.
        :param iterable inputs:
            An iterable of inputs for the generated targets. This can be an
            iterable of strings, tuples or dictionaries.
        :param mapping extra:
            A mapping of extra keyword arguments to be passed to the template.
        :param name: Must be either `None`, a string or a function.

            If `None` is given, the name of each target will be generated from
            the name of the template and an index.

            If a string is given, e.g. `foo`, the generated names will be
            `foo_0`, `foo_1`, etc.

            If a function is given, it must have the signature
            `f(idx, target)` where `idx` is the index and `target` is the
            :class:`AnonymousTarget` returned by the template. The function
            must return the name to assign to the target as a string.

        Any remaining keyword arguments will be passed directly to
        :func:`target_from_template` and thus override template-specified
        target options.
        """

        if extra is None:
            extra = {}

        if not (
            callable(template_func)
            and (
                hasattr(template_func, "__name__")
                or hasattr(template_func, "__class__")
            )
        ):
            raise ValueError(
                "Argument `template_func` must be a function or a callable "
                "class instance."
            )

        def template_namer(idx, target):
            if hasattr(template_func, "__name__"):
                name = template_func.__name__
            else:
                name = template_func.__class__.__name__
            return "{name}_{idx}".format(name=name, idx=idx)

        def string_namer(idx, target):
            return "{name}_{idx}".format(name=name, idx=idx)

        if name is None:
            name_func = template_namer
        elif isinstance(name, str):
            name_func = string_namer
        else:
            name_func = name

        targets = TargetList()
        for idx, args in enumerate(inputs):
            if isinstance(args, collections.abc.Mapping):
                template = template_func(**args, **extra)
            elif isinstance(args, collections.abc.Iterable) and not isinstance(
                args, str
            ):
                template = template_func(*args, **extra)
            else:
                template = template_func(args, **extra)

            target_name = name_func(idx, template)
            target = self.target_from_template(
                name=target_name, template=template, **kwargs
            )
            targets.append(target)
        return targets

    def include_path(self, path, namespace=None):
        """Include targets from another :class:`gwf.Workflow` into this workflow.

        See :func:`~gwf.Workflow.include`.
        """
        basedir, filename, obj = parse_path(path)
        other_workflow = load_workflow(basedir, filename, obj)
        self.include_workflow(other_workflow, namespace=namespace)

    def include_workflow(self, other_workflow, namespace=None):
        """Include targets from another :class:`gwf.Workflow` into this workflow.

        See :func:`~gwf.Workflow.include`.
        """
        if other_workflow.name is None and namespace is None:
            raise WorkflowError(
                "The included workflow has not been assigned a name. To "
                "include the workflow you must assign a name to the included "
                "workflow or set the namespace argument."
            )
        namespace_prefix = namespace or other_workflow.name
        if namespace_prefix == self.name:
            raise WorkflowError(
                "The included workflow has the same name as this workflow."
            )

        for target in other_workflow.targets.values():
            self._add_target(copy.deepcopy(target), namespace=namespace_prefix)

    def include(self, other_workflow, namespace=None):
        """Include targets from another :class:`gwf.Workflow` into this workflow.

        This method can be given either an :class:`gwf.Workflow` instance,
        a module or a path to a workflow file.

        If a module or path the workflow object to include will be determined
        according to the following rules:

        1. If a module object is given, the module must define an attribute
           named `gwf` containing a :class:`gwf.Workflow` object.
        2. If a path is given it must point to a file defining a module with an
           attribute named `gwf` containing a :class:`gwf.Workflow`
           object. If you want to include a workflow with another name you can
           specify the attribute name with a colon, e.g.::

                /some/path/workflow.py:myworkflow

           This will include all targets from the workflow `myworkflow`
           declared in the file `/some/path/workflow.py`.

        When a :class:`gwf.Workflow` instance has been obtained, all
        targets will be included directly into this workflow. To avoid name
        clashes the `namespace` argument must be provided. For example::

            workflow1 = Workflow()
            workflow1.target('TestTarget')

            workflow2 = Workflow()
            workflow2.target('TestTarget')

            workflow1.include(workflow2, namespace='wf1')

        The workflow now contains two targets named `TestTarget` (defined in
        `workflow2`) and `wf1.TestTarget` (defined in `workflow1`). The
        `namespace` parameter can be left out if the workflow to be included
        has been named::

            workflow1 = Workflow(name='wf1')
            workflow1.target('TestTarget')

            workflow2 = Workflow()
            workflow2.target('TestTarget')

            workflow1.include(workflow2)

        This yields the same result as before. The `namespace` argument can be
        used to override the specified name::

            workflow1 = Workflow(name='wf1')
            workflow1.target('TestTarget')

            workflow2 = Workflow()
            workflow2.target('TestTarget')

            workflow1.include(workflow2, namespace='foo')

        The workflow will now contain targets named `TestTarget` and
        `foo.TestTarget`.
        """
        if isinstance(other_workflow, Workflow):
            self.include_workflow(other_workflow, namespace=namespace)
        elif isinstance(other_workflow, str):
            self.include_path(other_workflow, namespace=namespace)
        elif inspect.ismodule(other_workflow):
            self.include_workflow(getattr(other_workflow, "gwf"), namespace=namespace)
        else:
            raise TypeError(
                "First argument must be either a string or a Workflow object."
            )

    def glob(self, pathname, *args, **kwargs):
        """Return a list of paths matching `pathname`.

        This method is equivalent to :func:`python:glob.glob`, but searches with
        relative paths will be performed relative to the working directory
        of the workflow.
        """
        if not os.path.isabs(pathname):
            pathname = os.path.join(self.working_dir, pathname)
        return _glob(pathname, *args, **kwargs)

    def iglob(self, pathname, *args, **kwargs):
        """Return an iterator which yields paths matching `pathname`.

        This method is equivalent to :func:`python:glob.iglob`, but searches with
        relative paths will be performed relative to the working directory
        of the workflow.
        """
        if not os.path.isabs(pathname):
            pathname = os.path.join(self.working_dir, pathname)
        return _iglob(pathname, *args, **kwargs)

    def shell(self, *args, **kwargs):
        """Return the output of a shell command.

        This method is equivalent to :func:`python:subprocess.check_output`, but
        automatically runs the command in a shell with the current working
        directory set to the working directory of the workflow.

        .. versionchanged:: 1.0

            This function no longer return a list of lines in the output, but a
            byte array with the output, exactly like
            :func:`python:subprocess.check_output`.
            You may specifically set *universal_newlines* to `True` to get a
            string with the output instead.
        """
        return subprocess.check_output(
            *args, shell=True, cwd=self.working_dir, **kwargs
        )

    def __repr__(self):
        return "{}(name={!r}, working_dir={!r})".format(
            self.__class__.__name__, self.name, self.working_dir
        )
from collections import Counter
from ..backends import Backend
from ..core import Graph, TargetStatus, get_status, schedule
from ..filtering import EndpointFilter, NameFilter, StatusFilter, filter_generic
from ..workflow import Workflow
import click
    TargetStatus.SHOULDRUN: "magenta",
    TargetStatus.SUBMITTED: "yellow",
    TargetStatus.RUNNING: "blue",
    TargetStatus.COMPLETED: "green",
}

STATUS_ORDER = (
    TargetStatus.SHOULDRUN,
    TargetStatus.SUBMITTED,
    TargetStatus.RUNNING,
    TargetStatus.COMPLETED,
)


def _status_str_to_enum(s):
    return TargetStatus[s.upper()]


def _status_strs_to_enums(iterable):
    return list(map(_status_str_to_enum, iterable))


def print_table(graph, targets, status_provider):
    targets = list(targets)

    name_col_width = max((len(target.name) for target in targets), default=0) + 4
    format_str = (
        "{name:<{name_col_width}}{status:<23}{percentage:>7.2%}"
        " [{num_shouldrun}/{num_submitted}/{num_running}/{num_completed}]"
    )

    for target in sorted(targets, key=lambda t: t.order):
        status = status_provider(target)
        color = STATUS_COLORS[status]

        deps = graph.dfs(target)
        deps_total = len(deps)

        def num_deps_with_status(status):
            return sum(1 for target in deps if status_provider(target) == status)

        num_shouldrun = num_deps_with_status(TargetStatus.SHOULDRUN)
        num_submitted = num_deps_with_status(TargetStatus.SUBMITTED)
        num_running = num_deps_with_status(TargetStatus.RUNNING)
        num_completed = num_deps_with_status(TargetStatus.COMPLETED)

        percentage = num_completed / deps_total

        line = format_str.format(
            name=target.name,
            status=click.style(status.name.lower(), fg=color),
            percentage=percentage,
            num_shouldrun=num_shouldrun,
            num_submitted=num_submitted,
            num_running=num_running,
            num_completed=num_completed,
            name_col_width=name_col_width,
        )
        click.echo(line)


def print_summary(backend, graph, targets):
    status_counts = Counter(get_status(target, backend=backend) for target in targets)
    click.echo("{:<15}{:>10}".format("total", len(targets)))
    for status in STATUS_ORDER:
        color = STATUS_COLORS[status]
        padded_name = "{:<15}".format(status.name.lower())
        click.echo(
            "{}{:>10}".format(click.style(padded_name, fg=color), status_counts[status])
        )


@click.command()
@click.argument("targets", nargs=-1)
@click.option("--endpoints", is_flag=True, default=False, help="Show only endpoints.")
@click.option(
    "--summary", is_flag=True, default=False, help="Only show summary statistics."
)
@click.option(
    "-s",
    "--status",
    type=click.Choice(["shouldrun", "submitted", "running", "completed"]),
    multiple=True,
)
@click.pass_obj
def status(obj, status, summary, endpoints, targets):
    """
    Show the status of targets.

    One target per line is shown. Each line contains the target name, the
    status of the target itself, and the percentage of dependencies of the
    target that are completed (including the target itself). That is, 100%
    means that the target and all of its dependencies have been completed.

    In square brackets, the number of targets that should run, have been
    submitted, are running, and completed targets are shown, respectively.

    The `-s/--status` flag can be applied multiple times to show targets that
    match either of the queries, e.g. `gwf status -s shouldrun -s completed`
    will show all targets that should run and all targets that are completed.

    The targets are shown in creation-order.
    """
    workflow = Workflow.from_config(obj)
    graph = Graph.from_targets(workflow.targets)
    backend_cls = Backend.from_config(obj)

    scheduled, _ = schedule(graph.endpoints(), graph=graph)

    def status_provider(target):
        return get_status(target, scheduled, backend)

    with backend_cls() as backend:
        filters = []
        if status:
            status = _status_strs_to_enums(status)
            filters.append(StatusFilter(status_provider=status_provider, status=status))
        if targets:
            filters.append(NameFilter(patterns=targets))
        if endpoints:

            filters.append(EndpointFilter(endpoints=graph.endpoints()))