task.py

from frozenlist2 import frozenlist

from pubtools.pulplib._impl import compat_attr as attr

from ..schema import load_schema
from .unit import Unit
from .common import PulpObject
from .attr import pulp_attrib


@attr.s(kw_only=True, frozen=True)
class Task(PulpObject):
    """Represents a Pulp task."""

    _SCHEMA = load_schema("task")

    id = pulp_attrib(type=str, pulp_field="task_id")
    """ID of this task (str)."""

    completed = pulp_attrib(default=None, type=bool)
    """True if this task has completed, successfully or otherwise.

    May be `None` if the state of this task is unknown.
    """

    succeeded = pulp_attrib(default=None, type=bool)
    """True if this task has completed successfully.

    May be `None` if the state of this task is unknown.
    """

    error_summary = pulp_attrib(default=None, type=str)
    """A summary of the reason for this task's failure (if any).

    This is a short string, generally a single line, suitable for display to users.
    The string includes the ID of the failed task.
    """

    error_details = pulp_attrib(default=None, type=str)
    """Detailed information for this task's failure (if any).

    This may be a multi-line string and may include technical information such as
    a Python backtrace generated by Pulp.

    ``error_details`` is a superset of the information available via ``error_summary``,
    so it is not necessary to display both.
    """

    tags = pulp_attrib(
        default=attr.Factory(frozenlist),
        type=list,
        converter=frozenlist,
        pulp_field="tags",
    )
    """The tags for this task.

    Typically includes info on the task's associated action and
    repo, such as:

    .. code-block:: python

        ["pulp:repository:rhel-7-server-rpms__7Server_x86_64",
         "pulp:action:publish"]
    """

    # TODO: is it a bug that this only allows a single repo ID??
    # Some tasks, like copy, involve multiple repos. We'll only include
    # the first repo ID from tags here... seems arbitrary.
    repo_id = pulp_attrib(type=str)
    """The ID of the repository associated with this task, otherwise None."""

    units = pulp_attrib(
        default=attr.Factory(frozenlist),
        type=list,
        pulp_field="result.units_successful",
        converter=frozenlist,
        pulp_py_converter=lambda raw: frozenlist(
            [Unit._from_task_data(x) for x in raw]
        ),
    )
    """Info on the units which were processed as part of this task
    (e.g. associated or unassociated).

    This is an iterable. Each element is an instance of
    :class:`~pubtools.pulplib.Unit` containing information on a processed
    unit.

    .. versionadded:: 1.5.0
    """

    units_data = pulp_attrib(
        default=attr.Factory(frozenlist),
        type=list,
        converter=frozenlist,
        pulp_field="result.units_successful",
    )
    """Info on the units which were processed as part of this task
    (e.g. associated or unassociated).

    This is a list. The list elements are the raw dicts as returned
    by Pulp. These should at least contain a 'type_id' and a 'unit_key'.

    .. deprecated:: 1.5.0
       Use :meth:`~pubtools.pulplib.Task.units` instead.
    """

    @repo_id.default
    def _repo_id_default(self):
        prefix = "pulp:repository:"
        for tag in self.tags or []:
            if tag.startswith(prefix):
                return tag[len(prefix) :]
        return None

    @succeeded.validator
    def _check_succeeded(self, _, value):
        if value and not self.completed:
            raise ValueError("Cannot have task with completed=False, succeeded=True")

    @classmethod
    def _data_to_init_args(cls, data):
        out = super(Task, cls)._data_to_init_args(data)

        state = data["state"]
        out["completed"] = state in ("finished", "error", "canceled", "skipped")
        out["succeeded"] = state in ("finished", "skipped")

        if state == "canceled":
            out["error_summary"] = "Pulp task [%s] was canceled" % data["task_id"]
            out["error_details"] = out["error_summary"]

        elif state == "error":
            out["error_summary"] = cls._error_summary(data)
            out["error_details"] = cls._error_details(data)

        return out

    @classmethod
    def _error_summary(cls, data):
        prefix = "Pulp task [%s] failed" % data["task_id"]
        error = data.get("error")
        if not error:
            return "%s: <unknown error>" % prefix
        return "%s: %s: %s" % (prefix, error["code"], error["description"])

    @classmethod
    def _error_details(cls, data):
        out = cls._error_summary(data)

        error = data.get("error")
        if not error:
            return out

        # Error looks like this:
        #
        # {
        #   'code': u'PLP0001',
        #   'data': {
        #     'message': 'a message'
        #   },
        #   'description': 'A general pulp exception occurred',
        #   'sub_errors': []
        # }
        #
        # See: https://docs.pulpproject.org/en/2.9/dev-guide/conventions/exceptions.html#error-details
        #
        # data can contain anything, or nothing.
        # It's only a convention that it often contains a message.
        #
        # sub_errors is currently ignored because I've never seen a non-empty
        # sub_errors yet.

        error_data = error.get("data") or {}
        messages = []

        # Message in a general exception
        if error_data.get("message"):
            messages.append(error_data["message"])

        # Some exceptions stash additional strings under details.errors
        if (error_data.get("details") or {}).get("errors"):
            error_messages = error_data["details"]["errors"]
            if isinstance(error_messages, list):
                messages.extend(error_messages)

        # Pulp docs refer to this as deprecated, but actually it's still
        # used and no alternative is provided.
        if data.get("traceback"):
            messages.append(data["traceback"])

        message = "\n".join(messages)
        if message:
            # message can have CRLF line endings in rare cases.
            message = message.replace("\r\n", "\n").strip()
            out = "%s:\n%s" % (out, _indent(message))

        return out


def _indent(text, level=2):
    spaces = " " * level
    return spaces + text.replace("\n", "\n" + spaces)