"""
State is the main currency in the Prefect platform. It is used to represent the current
status of a flow or task.

This module contains all Prefect state classes, all ultimately inheriting from the base State
class as follows:

![diagram of state inheritances](/state_inheritance_diagram.svg){.viz-padded}

Every run is initialized with the `Pending` state, meaning that it is waiting for
execution. During execution a run will enter a `Running` state. Finally, runs become `Finished`.
"""
import sys
import datetime
from typing import Any, Dict, List, Optional, Type, Mapping

import pendulum
import cloudpickle

import prefect
from prefect.engine.result import Result, NoResult


class State:
    """
    Base state class implementing the basic helper methods for checking state.

    **Note:** Each state-checking method (e.g., `is_failed()`) will also return `True`
    for all _subclasses_ of the parent state.  So, for example:
    ```python
    my_state = TriggerFailed()
    my_state.is_failed() # returns True

    another_state = Retrying()
    another_state.is_pending() # returns True
    ```

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
    """

    color = "#696969"

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        context: Dict[str, Any] = None,
        cached_inputs: Dict[str, Result] = None,
    ):
        self.message = message
        self.result = result if result is not None else NoResult
        self.context = context or dict()
        self.cached_inputs = cached_inputs or dict()  # type: Dict[str, Result]
        if "task_tags" in prefect.context:
            self.context.setdefault("tags", list(prefect.context.task_tags))

    def __repr__(self) -> str:
        if self.message is not None:
            return f'<{type(self).__name__}: "{self.message}">'
        else:
            return f"<{type(self).__name__}>"

    def __eq__(self, other: object) -> bool:
        """
        Equality depends on state type and data, but not message or context
        """
        if type(self) == type(other):
            assert isinstance(other, State)  # this assertion is here for MyPy only
            eq = self.result == other.result  # type: ignore
            for attr in self.__dict__:
                if attr.startswith("_") or attr in ["context", "message", "result"]:
                    continue
                eq &= getattr(self, attr, object()) == getattr(other, attr, object())
            return eq
        return False

    def __hash__(self) -> int:
        return id(self)

    def __getstate__(self) -> dict:
        data = self.__dict__

        # For 'Exception' results, we will check if it can be pickled successfully and
        # on failure convert the exception to a string instead. This allows the engine
        # to handle tasks with unpickable exceptions successfully. We must explicitly
        # ignore state signals here or we will get stuck in a recursive loop because
        # signals are exceptions that contain a state with `_result` set to the signal
        if isinstance(self.result, Exception) and not isinstance(
            self.result, prefect.engine.signals.PrefectStateSignal
        ):

            # Check for an exception during pickling of the 'Exception' type result
            try:
                cloudpickle.dumps(self.result)
            except Exception as exc:
                # Update the result in the dict without modifying the local object
                data = data.copy()
                new_result = data["_result"].copy()
                new_result.value = (
                    f"The following exception could not be pickled due to {exc!r}: "
                    f"{self.result!r}"
                )
                data["_result"] = new_result

        return data

    def __sizeof__(self) -> int:
        return super().__sizeof__() + sys.getsizeof(self.result, 0)

    @property
    def result(self) -> Any:
        return getattr(self._result, "value", self._result)

    @result.setter
    def result(self, value: Any) -> None:
        if isinstance(value, Result):
            self._result = value
        else:
            self._result = Result(value=value)

    def load_result(self, result: Result = None) -> "State":
        """
        Given another Result instance, uses the current Result's `location` to create a fully
        hydrated `Result` using the logic of the provided result.  This method is mainly
        intended to be used by `TaskRunner` methods to hydrate deserialized Cloud results into
        fully functional `Result` instances.

        Args:
            - result (Result): the result instance to hydrate with `self.location`

        Returns:
            - State: the current state with a fully hydrated Result attached
        """
        if self.is_mapped():
            self.map_states = [
                s.load_result(result) if s is not None else None
                for s in self.map_states  # type: ignore
            ]
            if self.map_states:
                self.result = [
                    s.result if s is not None else None for s in self.map_states
                ]
            return self

        result_reader = result or self._result

        known_location = self._result.location  # type: ignore

        if self._result.value is None and known_location is not None:  # type: ignore
            self._result = result_reader.read(known_location)  # type: ignore
        return self

    def load_cached_results(
        self, results: Mapping[str, Optional[Result]] = None
    ) -> "State":
        """
        Given another Result instance, uses the current Result's `location` to create a fully
        hydrated `Result` using the logic of the provided result.  This method is mainly
        intended to be used by `TaskRunner` methods to hydrate deserialized Cloud results into
        fully functional `Result` instances.

        Args:
            - results (Dict[str, Result]): a dictionary of result instances to hydrate
                `self.cached_inputs` with

        Returns:
            - State: the current state with a fully hydrated Result attached
        """
        results = results or dict()

        result_readers = {
            key: results.get(key, result) for key, result in self.cached_inputs.items()
        }  # type: ignore

        loaded_inputs = {}

        for key, res in self.cached_inputs.items():
            known_location = res.location or getattr(
                result_readers[key], "location", None
            )
            if res.value is None and known_location is not None:
                loaded_inputs[key] = result_readers[key].read(known_location)  # type: ignore
            else:
                loaded_inputs[key] = res

        self.cached_inputs = loaded_inputs

        return self

    @classmethod
    def children(
        cls, include_self: bool = False, names_only: bool = False
    ) -> "List[Type[State]]":
        """
        Helper method for retrieving all possible child states of this state.

        Args:
            - include_self (bool, optional): whether to include the calling state in the return
                values; defaults to `False`
            - names_only (bool, optional): whether to only return the string names of the states;
                defaults to `False`, in which case the actual classes are returned

        Returns:
            - list: a (possibly empty) list of states or state names
        """
        children = []
        for state in cls.__subclasses__():
            # hide "private" state types
            if not state.__name__.startswith("_"):
                children.append(state)
            children.extend(state.children())
        if include_self:
            children += [cls]
        if names_only:
            return [s.__name__ for s in children]  # type: ignore
        return children

    @classmethod
    def parents(
        cls, include_self: bool = False, names_only: bool = False
    ) -> "List[Type[State]]":
        """
        Helper method for retrieving all possible parent states of this state.

        Args:
            - include_self (bool, optional): whether to include the calling state in the return
                values; defaults to `False`
            - names_only (bool, optional): whether to only return the string names of the states;
                defaults to `False`, in which case the actual classes are returned

        Returns:
            - list: a (possibly empty) list of states or state names
        """
        parents = []
        for state in cls.mro():
            if state in [object, cls]:
                continue

            # hide "private" state types
            if not state.__name__.startswith("_"):
                parents.append(state)
        if include_self:
            parents += [cls]
        if names_only:
            return [s.__name__ for s in parents]  # type: ignore
        return parents

    def is_pending(self) -> bool:
        """
        Checks if the state is currently in a pending state

        Returns:
            - bool: `True` if the state is pending, `False` otherwise
        """

        return isinstance(self, Pending)

    def is_queued(self) -> bool:
        """
        Checks if the state is currently in a queued state

        Returns:
            - bool: `True` if the state is queued, `False` otherwise
        """

        return isinstance(self, Queued)

    def is_retrying(self) -> bool:
        """
        Checks if the state is currently in a retrying state

        Returns:
            - bool: `True` if the state is retrying, `False` otherwise
        """

        return isinstance(self, Retrying)

    def is_running(self) -> bool:
        """
        Checks if the state is currently in a running state

        Returns:
            - bool: `True` if the state is running, `False` otherwise
        """
        return isinstance(self, Running)

    def is_cached(self) -> bool:
        """
        Checks if the state is currently in a Cached state

        Returns:
            - bool: `True` if the state is Cached, `False` otherwise
        """
        return isinstance(self, Cached)

    def is_finished(self) -> bool:
        """
        Checks if the state is currently in a finished state

        Returns:
            - bool: `True` if the state is finished, `False` otherwise
        """
        return isinstance(self, Finished)

    def is_looped(self) -> bool:
        """
        Checks if the state is currently in a looped state

        Returns:
            - bool: `True` if the state is looped, `False` otherwise
        """
        return isinstance(self, Looped)

    def is_scheduled(self) -> bool:
        """
        Checks if the state is currently in a scheduled state, which includes retrying.

        Returns:
            - bool: `True` if the state is skipped, `False` otherwise
        """
        return isinstance(self, (Scheduled, Queued))

    def is_submitted(self) -> bool:
        """
        Checks if the state is currently in a submitted state.

        Returns:
            - bool: `True` if the state is submitted, `False` otherwise
        """
        return isinstance(self, Submitted)

    def is_skipped(self) -> bool:
        """
        Checks if the state is currently in a skipped state

        Returns:
            - bool: `True` if the state is skipped, `False` otherwise
        """
        return isinstance(self, Skipped)

    def is_successful(self) -> bool:
        """
        Checks if the state is currently in a successful state

        Returns:
            - bool: `True` if the state is successful, `False` otherwise
        """
        return isinstance(self, Success)

    def is_failed(self) -> bool:
        """
        Checks if the state is currently in a failed state

        Returns:
            - bool: `True` if the state is failed, `False` otherwise
        """
        return isinstance(self, Failed)

    def is_mapped(self) -> bool:
        """
        Checks if the state is currently in a mapped state

        Returns:
            - bool: `True` if the state is mapped, `False` otherwise
        """
        return isinstance(self, Mapped)

    def is_meta_state(self) -> bool:
        """
        Checks if the state is a meta state that wraps another state

        Returns:
            - bool: `True` if the state is a meta state, `False` otherwise
        """
        return isinstance(self, _MetaState)

    @staticmethod
    def deserialize(json_blob: dict) -> "State":
        """
        Deserializes the state from a dict.

        Args:
            - json_blob (dict): the JSON representing the serialized state
        """
        from prefect.serialization.state import StateSchema

        state = StateSchema().load(json_blob)
        return state

    def serialize(self) -> dict:
        """
        Serializes the state to a dict.

        Returns:
            - dict: a JSON representation of the state
        """
        from prefect.serialization.state import StateSchema

        json_blob = StateSchema().dump(self)
        return json_blob


# -------------------------------------------------------------------
# Pending States
# -------------------------------------------------------------------


class Pending(State):
    """
    Base Pending state; default state for new tasks.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#7ebdff"

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        cached_inputs: Dict[str, Result] = None,
        context: Dict[str, Any] = None,
    ):
        super().__init__(
            message=message, result=result, context=context, cached_inputs=cached_inputs
        )


class Scheduled(Pending):
    """
    Pending state indicating the object has been scheduled to run.

    Scheduled states have a `start_time` that indicates when they are scheduled to run.
    Only scheduled states have this property; this is important because non-Python systems
    identify scheduled states by the presence of this property.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - start_time (datetime): time at which the task is scheduled to run
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#ffab00"
    start_time: Optional[datetime.datetime]

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        start_time: datetime.datetime = None,
        cached_inputs: Dict[str, Result] = None,
        context: Dict[str, Any] = None,
    ):
        super().__init__(
            message=message, result=result, cached_inputs=cached_inputs, context=context
        )
        self.start_time = pendulum.instance(start_time or pendulum.now("utc"))
        run_count = prefect.context.get("task_run_count")
        if run_count is not None:
            self.context.update(task_run_count=run_count)


class Paused(Scheduled):
    """
    Paused state for tasks. This allows manual intervention or pausing for a set amount of
    time.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - start_time (datetime): time at which the task is scheduled to resume; defaults
            to 10 years from now if not provided.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#99a8e8"

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        start_time: datetime.datetime = None,
        cached_inputs: Dict[str, Result] = None,
        context: Dict[str, Any] = None,
    ):

        super().__init__(
            message=message,
            result=result,
            start_time=start_time,
            cached_inputs=cached_inputs,
            context=context,
        )

        # override default logic to set start_time = now();
        # have indefinite start_time instead
        if start_time is None:
            self.start_time = None


class _MetaState(State):
    """
    Meta states are used to provide meta-information about other states. For example,
    the Submitted state wraps another state to show that it has been handled. The Queued
    state wraps states to show that they have been queued for execution.

    This parent class should NOT generally be used, but allows all meta states to be
    easily identified.
    """

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        state: State = None,
        context: Dict[str, Any] = None,
        cached_inputs: Dict[str, Result] = None,
    ):
        super().__init__(
            message=message, result=result, context=context, cached_inputs=cached_inputs
        )
        self.state = state

    @property
    def state(self) -> Optional[State]:
        return self._state

    @state.setter
    def state(self, val: Optional[State]) -> None:
        while isinstance(val, State) and val.is_meta_state():
            val = val.state  # type: ignore
        self._state = val


class ClientFailed(_MetaState):
    """
    The `ClientFailed` state is used to indicate that the Prefect Client failed
    to set a task run state, and thus this task run should exit, without triggering any
    downstream task runs.

    The `ClientFailed` state should be initialized with another state, which it wraps. The
    wrapped state is the state which the client attempted to set in the database, but failed to
    for some reason.

    Args:
        - message (string): a message for the state.
        - result (Any, optional): Defaults to `None`.
        - state (State): the `State` state that the task run ended in
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible

    """

    color = "#eb0000"


class Submitted(_MetaState):
    """
    The `Submitted` state is used to indicate that another state, usually a `Scheduled`
    state, has been handled. For example, if a task is in a `Retrying` state, then at
    the appropriate time it may be put into a `Submitted` state referencing the
    `Retrying` state. This communicates to the system that the retry has been handled,
    without losing the information contained in the `Retry` state.

    The `Submitted` state should be initialized with another state, which it wraps. The
    wrapped state is extracted at the beginning of a task run.

    Args:
        - message (string): a message for the state.
        - result (Any, optional): Defaults to `None`.
        - state (State): the `State` state that has been marked as "submitted".
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible

    """

    color = "#ffdf5d"


class Queued(_MetaState):
    """
    The `Queued` state is used to indicate that another state could not transition to a
    `Running` state for some reason, often a lack of available resources.

    The `Queued` state should be initialized with another state, which it wraps. The
    wrapped state is extracted at the beginning of a task run.

    Args:
        - message (string): a message for the state.
        - result (Any, optional): Defaults to `None`.
        - state (State): the `State` state that has been marked as
            "queued".
        - start_time (datetime): a time the state is queued until. Defaults to `now`.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible

    """

    color = "#ffea7f"

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        state: State = None,
        start_time: datetime.datetime = None,
        context: Dict[str, Any] = None,
        cached_inputs: Dict[str, Result] = None,
    ):
        super().__init__(
            message=message,
            result=result,
            state=state,
            context=context,
            cached_inputs=cached_inputs,
        )
        self.start_time = start_time or pendulum.now("utc")


class Resume(Scheduled):
    """
    Resume state indicating the object can resume execution (presumably from a `Paused` state).

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - start_time (datetime): time at which the task is scheduled to run
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#f58c0c"


class Retrying(Scheduled):
    """
    Pending state indicating the object has been scheduled to be retried.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - start_time (datetime): time at which the task is scheduled to be retried
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
        - run_count (int): The number of runs that had been attempted at the time of this
            Retry. Defaults to the value stored in context under "task_run_count" or 1,
            if that value isn't found.
    """

    color = "#f66a0a"

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        start_time: datetime.datetime = None,
        cached_inputs: Dict[str, Result] = None,
        context: Dict[str, Any] = None,
        run_count: int = None,
    ):
        super().__init__(
            result=result,
            message=message,
            start_time=start_time,
            cached_inputs=cached_inputs,
            context=context,
        )
        if run_count is None:
            run_count = prefect.context.get("task_run_count", 1)
        assert run_count is not None  # mypy assert
        self.run_count = run_count  # type: int


# -------------------------------------------------------------------
# Running States
# -------------------------------------------------------------------


class Running(State):
    """
    Base running state. Indicates that a task is currently running.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#3d67ff"


class Cancelling(Running):
    """
    State indicating that a previously running flow run is in the process of
    cancelling, but still may have tasks running.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#E8E8E8"


# -------------------------------------------------------------------
# Finished States
# -------------------------------------------------------------------


class Finished(State):
    """
    Base finished state. Indicates when a class has reached some form of completion.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#003ccb"


class Looped(Finished):
    """
    Finished state indicating one successful run of a looped task - if a Task is in this state,
    it will run the next iteration of the loop immediately after.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - loop_count (int): The iteration number of the looping task.
            Defaults to the value stored in context under "task_loop_count" or 1,
            if that value isn't found.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#003ccb"

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        loop_count: int = None,
        context: Dict[str, Any] = None,
        cached_inputs: Dict[str, Result] = None,
    ):
        super().__init__(
            result=result, message=message, context=context, cached_inputs=cached_inputs
        )
        if loop_count is None:
            loop_count = prefect.context.get("task_loop_count", 1)
        assert loop_count is not None  # mypy assert
        self.loop_count = loop_count  # type: int


class Success(Finished):
    """
    Finished state indicating success.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#28a745"


class Cached(Success):
    """
    Cached, which represents a Task whose outputs have been cached.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the
            state, which will be cached.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - cached_parameters (dict): Defaults to `None`
        - cached_result_expiration (datetime): The time at which this cache
            expires and can no longer be used. Defaults to `None`
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
        - hashed_inputs (Dict[str, str], optional): a string hash of a dictionary of inputs
    """

    color = "#34d058"

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        cached_inputs: Dict[str, Result] = None,
        cached_parameters: Dict[str, Any] = None,
        cached_result_expiration: datetime.datetime = None,
        context: Dict[str, Any] = None,
        hashed_inputs: Dict[str, str] = None,
    ):
        super().__init__(
            message=message, result=result, context=context, cached_inputs=cached_inputs
        )
        self.hashed_inputs = hashed_inputs
        self.cached_parameters = cached_parameters  # type: Optional[Dict[str, Any]]
        if cached_result_expiration is not None:
            cached_result_expiration = pendulum.instance(cached_result_expiration)
        self.cached_result_expiration = (
            cached_result_expiration
        )  # type: Optional[datetime.datetime]


class Mapped(Success):
    """
    State indicated this task was mapped over, and all mapped tasks were _submitted_ successfully.
    Note that this does _not_ imply the individual mapped tasks were successful, just that they
    have been submitted.

    You can not set the `result` of a Mapped state; it is determined by the results of its
    children states.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `[]`. A data payload for the state.
        - map_states (List): A list containing the states of any "children" of this task. When
            a task enters a Mapped state, it indicates that it has dynamically created copies
            of itself to map its operation over its inputs. Those copies are the children.
        - n_map_states (int, optional): the number of tasks that were mapped; if not provided,
            the value of `len(map_states)` is used
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#003ccb"

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        map_states: List[State] = None,
        context: Dict[str, Any] = None,
        cached_inputs: Dict[str, Result] = None,
        n_map_states: int = None,
    ):
        super().__init__(
            message=message, result=result, context=context, cached_inputs=cached_inputs
        )
        self.map_states = map_states or []  # type: List[State]
        self.n_map_states = n_map_states  # type: ignore

    @property
    def n_map_states(self) -> int:
        return self._n_map_states or len(self.map_states)

    @n_map_states.setter
    def n_map_states(self, val: Optional[int]) -> None:
        self._n_map_states = val


class Cancelled(Finished):
    """
    Finished state indicating that a user cancelled the flow run manually, mid-run.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#bdbdbd"


class Failed(Finished):
    """
    Finished state indicating failure.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#eb0000"

    def __init__(
        self,
        message: str = None,
        result: Any = None,
        cached_inputs: Dict[str, Result] = None,
        context: Dict[str, Any] = None,
    ):
        super().__init__(
            message=message, result=result, context=context, cached_inputs=cached_inputs
        )


class TimedOut(Failed):
    """
    Finished state indicating failure due to execution timeout.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#ff4e33"


class TriggerFailed(Failed):
    """
    Finished state indicating failure due to trigger.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#ff5131"


class ValidationFailed(Failed):
    """
    Finished stated indicating failure due to failed result validation.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#ff5131"


class Skipped(Success):
    """
    Finished state indicating success on account of being skipped.

    Args:
        - message (str or Exception, optional): Defaults to `None`. A message about the
            state, which could be an `Exception` (or [`Signal`](signals.html)) that caused it.
        - result (Any, optional): Defaults to `None`. A data payload for the state.
        - cached_inputs (dict, optional, DEPRECATED): A dictionary of input keys to fully hydrated
            `Result`s. Used / set if the Task requires retries.
        - context (dict, optional): A dictionary of execution context information; values
            should be JSON compatible
    """

    color = "#62757f"

    # note: this does not allow setting "cached" as Success states do
    def __init__(
        self,
        message: str = None,
        result: Any = None,
        context: Dict[str, Any] = None,
        cached_inputs: Dict[str, Result] = None,
    ):
        super().__init__(
            message=message, result=result, context=context, cached_inputs=cached_inputs
        )
