community.docker/plugins/module_utils/_api/errors.py

# This code is part of the Ansible collection community.docker, but is an independent component.
# This particular file, and this file only, is based on the Docker SDK for Python (https://github.com/docker/docker-py/)
#
# Copyright (c) 2016-2022 Docker, Inc.
#
# It is licensed under the Apache 2.0 license (see LICENSES/Apache-2.0.txt in this collection)
# SPDX-License-Identifier: Apache-2.0

# Note that this module util is **PRIVATE** to the collection. It can have breaking changes at any time.
# Do not use this from other collections or standalone plugins/modules!

from __future__ import annotations

import typing as t

from ansible.module_utils.common.text.converters import to_text

from ._import_helper import HTTPError as _HTTPError

if t.TYPE_CHECKING:
    from requests import Response


class DockerException(Exception):
    """
    A base class from which all other exceptions inherit.

    If you want to catch all errors that the Docker SDK might raise,
    catch this base exception.
    """


def create_api_error_from_http_exception(e: _HTTPError) -> t.NoReturn:
    """
    Create a suitable APIError from requests.exceptions.HTTPError.
    """
    response = e.response
    try:
        explanation = response.json()["message"]
    except ValueError:
        explanation = to_text((response.content or "").strip())
    cls = APIError
    if response.status_code == 404:
        if explanation and (
            "No such image" in str(explanation)
            or "not found: does not exist or no pull access" in str(explanation)
            or "repository does not exist" in str(explanation)
            or "was found but does not match the specified platform" in str(explanation)
            or "failed to resolve reference" in str(explanation)
        ):
            cls = ImageNotFound
        else:
            cls = NotFound
    raise cls(e, response=response, explanation=explanation) from e


class APIError(_HTTPError, DockerException):
    """
    An HTTP error from the API.
    """

    def __init__(
        self,
        message: str | Exception,
        response: Response | None = None,
        explanation: str | None = None,
    ) -> None:
        # requests 1.2 supports response as a keyword argument, but
        # requests 1.1 does not
        super().__init__(message)
        self.response = response
        self.explanation = explanation or ""

    def __str__(self) -> str:
        message = super().__str__()

        if self.is_client_error():
            message = f"{self.response.status_code} Client Error for {self.response.url}: {self.response.reason}"

        elif self.is_server_error():
            message = f"{self.response.status_code} Server Error for {self.response.url}: {self.response.reason}"

        if self.explanation:
            message = f'{message} ("{self.explanation}")'

        return message

    @property
    def status_code(self) -> int | None:
        if self.response is not None:
            return self.response.status_code
        return None

    def is_error(self) -> bool:
        return self.is_client_error() or self.is_server_error()

    def is_client_error(self) -> bool:
        if self.status_code is None:
            return False
        return 400 <= self.status_code < 500

    def is_server_error(self) -> bool:
        if self.status_code is None:
            return False
        return 500 <= self.status_code < 600


class NotFound(APIError):
    pass


class ImageNotFound(NotFound):
    pass


class InvalidVersion(DockerException):
    pass


class InvalidRepository(DockerException):
    pass


class InvalidConfigFile(DockerException):
    pass


class InvalidArgument(DockerException):
    pass


class DeprecatedMethod(DockerException):
    pass


class TLSParameterError(DockerException):
    def __init__(self, msg: str) -> None:
        self.msg = msg

    def __str__(self) -> str:
        return self.msg + (
            ". TLS configurations should map the Docker CLI "
            "client configurations. See "
            "https://docs.docker.com/engine/articles/https/ "
            "for API details."
        )


class NullResource(DockerException, ValueError):
    pass


class ContainerError(DockerException):
    """
    Represents a container that has exited with a non-zero exit code.
    """

    def __init__(
        self,
        container: str,
        exit_status: int,
        command: list[str],
        image: str,
        stderr: str | None,
    ):
        self.container = container
        self.exit_status = exit_status
        self.command = command
        self.image = image
        self.stderr = stderr

        err = f": {stderr}" if stderr is not None else ""
        msg = f"Command '{command}' in image '{image}' returned non-zero exit status {exit_status}{err}"

        super().__init__(msg)


class StreamParseError(RuntimeError):
    def __init__(self, reason: Exception) -> None:
        self.msg = reason


class BuildError(DockerException):
    def __init__(self, reason: str, build_log: str) -> None:
        super().__init__(reason)
        self.msg = reason
        self.build_log = build_log


class ImageLoadError(DockerException):
    pass


def create_unexpected_kwargs_error(name: str, kwargs: dict[str, t.Any]) -> TypeError:
    quoted_kwargs = [f"'{k}'" for k in sorted(kwargs)]
    text = [f"{name}() "]
    if len(quoted_kwargs) == 1:
        text.append("got an unexpected keyword argument ")
    else:
        text.append("got unexpected keyword arguments ")
    text.append(", ".join(quoted_kwargs))
    return TypeError("".join(text))


class MissingContextParameter(DockerException):
    def __init__(self, param: str) -> None:
        self.param = param

    def __str__(self) -> str:
        return f"missing parameter: {self.param}"


class ContextAlreadyExists(DockerException):
    def __init__(self, name: str) -> None:
        self.name = name

    def __str__(self) -> str:
        return f"context {self.name} already exists"


class ContextException(DockerException):
    def __init__(self, msg: str) -> None:
        self.msg = msg

    def __str__(self) -> str:
        return self.msg


class ContextNotFound(DockerException):
    def __init__(self, name: str) -> None:
        self.name = name

    def __str__(self) -> str:
        return f"context '{self.name}' not found"


class MissingRequirementException(DockerException):
    def __init__(
        self, msg: str, requirement: str, import_exception: ImportError | str
    ) -> None:
        self.msg = msg
        self.requirement = requirement
        self.import_exception = import_exception

    def __str__(self) -> str:
        return self.msg