Source code for taskflow.deciders

# -*- coding: utf-8 -*-

#    Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import six

from taskflow.utils import misc


[docs]class Depth(misc.StrEnum):
    """Enumeration of decider(s) *area of influence*."""

    ALL = 'ALL'
    """
    **Default** decider depth that affects **all** successor atoms (including
    ones that are in successor nested flows).
    """

    FLOW = 'FLOW'
    """
    Decider depth that affects **all** successor tasks in the **same**
    flow (it will **not** affect tasks/retries that are in successor
    nested flows).

    .. warning::

       While using this kind we are allowed to execute successors of
       things that have been ignored (for example nested flows and the
       tasks they contain), this may result in symbol lookup errors during
       running, user beware.
    """

    NEIGHBORS = 'NEIGHBORS'
    """
    Decider depth that affects only **next** successor tasks (and does
    not traverse past **one** level of successor tasks).

    .. warning::

       While using this kind we are allowed to execute successors of
       things that have been ignored (for example nested flows and the
       tasks they contain), this may result in symbol lookup errors during
       running, user beware.
    """

    ATOM = 'ATOM'
    """
    Decider depth that affects only **targeted** atom (and does
    **not** traverse into **any** level of successor atoms).

    .. warning::

       While using this kind we are allowed to execute successors of
       things that have been ignored (for example nested flows and the
       tasks they contain), this may result in symbol lookup errors during
       running, user beware.
    """

[docs]    @classmethod
    def translate(cls, desired_depth):
        """Translates a string into a depth enumeration."""
        if isinstance(desired_depth, cls):
            # Nothing to do in the first place...
            return desired_depth
        if not isinstance(desired_depth, six.string_types):
            raise TypeError("Unexpected desired depth type, string type"
                            " expected, not %s" % type(desired_depth))
        try:
            return cls(desired_depth.upper())
        except ValueError:
            pretty_depths = sorted([a_depth.name for a_depth in cls])
            raise ValueError("Unexpected decider depth value, one of"
                             " %s (case-insensitive) is expected and"
                             " not '%s'" % (pretty_depths, desired_depth))


# Depth area of influence order (from greater influence to least).
#
# Order very much matters here...
_ORDERING = tuple([
    Depth.ALL, Depth.FLOW, Depth.NEIGHBORS, Depth.ATOM,
])


[docs]def pick_widest(depths):
    """Pick from many depths which has the **widest** area of influence."""
    return _ORDERING[min(_ORDERING.index(d) for d in depths)]