path: root/src/pybind/mgr/object_format.py

# object_format.py provides types and functions for working with
# requested output formats such as JSON, YAML, etc.
"""tools for writing formatting-friendly mgr module functions

Currently, the ceph mgr code in python is most commonly written by adding mgr
modules and corresponding classes and then adding methods to those classes that
are decorated using `@CLICommand` from  `mgr_module.py`.  These methods (that
will be called endpoints subsequently) then implement the logic that is
executed when the mgr receives a command from a client.  These endpoints are
currently responsible for forming a response tuple of (int, str, str) where the
int represents a return value (error code) and the first string the "body" of
the response. The mgr supports a generic `format` parameter (`--format` on the
ceph cli) that each endpoint must then explicitly handle. At the time of this
writing, many endpoints do not handle alternate formats and are each
implementing formatting/serialization of values in various different ways.

The `object_format` module aims to make the process of writing endpoint
functions easier, more consistent, and (hopefully) better documented.  At the
highest level, the module provides a new decorator `Responder` that must be
placed below the `CLICommand` decorator (so that it decorates the endpoint
before `CLICommand`). This decorator helps automatically convert Python objects
to response tuples expected by the manager, while handling the `format`
parameter automatically.

In addition to the decorator the module provides a few other types and methods
that intended to interoperate with the decorator and make small customizations
and error handling easier.

== Using Responder ==

The simple and intended way to use the decorator is as follows:
    @CLICommand("command name", perm="r")
    Responder()
    def create_something(self, name: str) -> Dict[str, str]:
        ...  # implementation
        return {"name": name, "id": new_id}

In this case the `create_something` method return a python dict,
and does not return a response tuple directly. Instead, the
dict is converted to either JSON or YAML depending on what the
client requested. Assuming no exception is raised by the
implementation then the response code is always zero (success).

The object_format module provides an exception type `ErrorResponse`
that assists in returning "clean" error conditions to the client.
Extending the previous example to use this exception:
    @CLICommand("command name", perm="r")
    Responder()
    def create_something(self, name: str) -> Dict[str, str]:
        try:
           ...  # implementation
           return {"name": name, "id": new_id}
        except KeyError as kerr:
           # explicitly set the return value to ENOENT for KeyError
           raise ErrorResponse.wrap(kerr, return_value=-errno.ENOENT)
        except (BusinessLogcError, OSError) as err:
           # return value is based on err when possible
           raise ErrorResponse.wrap(err)

Most uses of ErrorResponse are expected to use the `wrap` classmethod,
as it will aid in the handling of an existing exception but `ErrorResponse`
can be used directly too.

== Customizing Response Formatting ==

The `Responder` is built using two additional mid-layer types. The
`ObjectFormatAdapter` and the `ReturnValueAdapter` by default. These types
implement the `CommonFormatter` protocol and `ReturnValueProvider` protocols
respectively. Most cases will not need to customize the `ReturnValueAdapter` as
returning zero on success is expected.  However, if there's a need to return a
non-zero error code outside of an exception, you can add the `mgr_return_value`
function to the returned type of the endpoint function - causing it to meet the
`ReturnValueProvider` protocol. Whatever integer that function returns will
then be used in the response tuple.

The `ObjectFormatAdapter` can operate in two modes. By default, any type
returned from the endpoint function will be checked for a `to_simplified`
method (the type matches the SimpleDataProvider` protocol) and if it exists
the method will be called and the result serialized. Example:
    class CoolStuff:
       def __init__(self, temperature: int, quantity: int) -> None:
           self.temperature = temperature
           self.quantity = quantity
      def to_simplified(self) -> Dict[str, int]:
         return {"temp": self.temperature, "qty": self.quantity}

    @CLICommand("command name", perm="r")
    Responder()
    def create_something_cool(self) -> CoolStuff:
       cool_stuff: CoolStuff = self._make_cool_stuff()  # implementation
       return cool_stuff

In order to serialize the result, the object returned from the wrapped
function must provide the `to_simplified` method (or the compatibility methods,
see below) or already be a "simplified type". Valid types include lists and
dicts that contain other lists and dicts and ints, strs, bools -- basic objects
that can be directly converted to json (via json.dumps) without any additional
conversions. The `to_simplified` method must always return such types.

To be compatible with many existing types in the ceph mgr codebase one can pass
`compatible=True` to the `ObjectFormatAdapter`. If the type provides a
`to_json` and/or `to_yaml` method that returns basic python types (dict, list,
str, etc...) but *not* already serialized JSON or YAML this flag can be
enabled. Note that Responder takes as an argument any callable that returns a
`CommonFormatter`. In this example below we enable the flag using
`functools.partial`:
    class MyExistingClass:
      def to_json(self) -> Dict[str, Any]:
         return {"name": self.name, "height": self.height}

    @CLICommand("command name", perm="r")
    Responder(functools.partial(ObjectFormatAdapter, compatible=True))
    def create_an_item(self) -> MyExistingClass:
       item: MyExistingClass = self._new_item()  # implementation
       return item


For cases that need to return xml or plain text formatted responses one can
create a new class that matches the `CommonFormatter` protocol (provides a
valid_formats method) and one or more `format_x` method where x is the name of
a format ("json", "yaml", "xml", "plain", etc...).
    class MyCustomFormatAdapter:
       def __init__(self, obj_to_format: Any) -> None:
           ...
       def valid_formats(self) -> Iterable[str]:
           ...
       def format_json(self) -> str:
           ...
       def format_xml(self) -> str:
           ...


Of course, the Responder itself can be used as a base class and aspects of the
Responder altered for specific use cases. Inheriting from `Responder` and
customizing it is an exercise left for those brave enough to read the code in
`object_format.py` :-).
"""

import enum
import errno
import json
import sys

from functools import wraps
from typing import (
    Any,
    Callable,
    Dict,
    Iterable,
    List,
    Optional,
    TYPE_CHECKING,
    Tuple,
    Type,
    TypeVar,
    Union,
)

import yaml

# this uses a version check as opposed to a try/except because this
# form makes mypy happy and try/except doesn't.
if sys.version_info >= (3, 8):
    from typing import Protocol
elif TYPE_CHECKING:
    # typing_extensions will not be available for the real mgr server
    from typing_extensions import Protocol
else:
    # fallback type that is acceptable to older python on prod. builds
    class Protocol:  # type: ignore
        pass

from mgr_module import HandlerFuncType


DEFAULT_JSON_INDENT: int = 2


class Format(str, enum.Enum):
    plain = "plain"
    json = "json"
    json_pretty = "json-pretty"
    yaml = "yaml"
    xml_pretty = "xml-pretty"
    xml = "xml"


# SimpleData is a type alias for Any unless we can determine the
# exact set of subtypes we want to support. But it is explicit!
SimpleData = Any


class SimpleDataProvider(Protocol):
    def to_simplified(self) -> SimpleData:
        """Return a simplified representation of the current object.
        The simplified representation should be trivially serializable.
        """
        ...  # pragma: no cover


class JSONDataProvider(Protocol):
    def to_json(self) -> Any:
        """Return a python object that can be serialized into JSON.
        This function does _not_ return a JSON string.
        """
        ...  # pragma: no cover


class YAMLDataProvider(Protocol):
    def to_yaml(self) -> Any:
        """Return a python object that can be serialized into YAML.
        This function does _not_ return a string of YAML.
        """
        ...  # pragma: no cover


class JSONFormatter(Protocol):
    def format_json(self) -> str:
        """Return a JSON formatted representation of an object."""
        ...  # pragma: no cover


class YAMLFormatter(Protocol):
    def format_yaml(self) -> str:
        """Return a JSON formatted representation of an object."""
        ...  # pragma: no cover


class ReturnValueProvider(Protocol):
    def mgr_return_value(self) -> int:
        """Return an integer value to provide the Ceph MGR with a error code
        for the MGR's response tuple. Zero means success. Return an negative
        errno otherwise.
        """
        ...  # pragma: no cover


class CommonFormatter(Protocol):
    """A protocol that indicates the type is a formatter for multiple
    possible formats.
    """

    def valid_formats(self) -> Iterable[str]:
        """Return the names of known valid formats."""
        ...  # pragma: no cover


# The _is_name_of_protocol_type functions below are here because the production
# builds of the ceph manager are lower than python 3.8 and do not have
# typing_extensions available in the resulting images. This means that
# runtime_checkable is not available and isinstance can not be used with a
# protocol type.  These could be replaced by isinstance in a later version of
# python.  Note that these functions *can not* be methods of the protocol types
# for neatness - including methods on the protocl types makes mypy consider
# those methods as part of the protcol & a required method. Using decorators
# did not change that - I checked.


def _is_simple_data_provider(obj: SimpleDataProvider) -> bool:
    """Return true if obj is usable as a SimpleDataProvider."""
    return callable(getattr(obj, 'to_simplified', None))


def _is_json_data_provider(obj: JSONDataProvider) -> bool:
    """Return true if obj is usable as a JSONDataProvider."""
    return callable(getattr(obj, 'to_json', None))


def _is_yaml_data_provider(obj: YAMLDataProvider) -> bool:
    """Return true if obj is usable as a YAMLDataProvider."""
    return callable(getattr(obj, 'to_yaml', None))


def _is_return_value_provider(obj: ReturnValueProvider) -> bool:
    """Return true if obj is usable as a YAMLDataProvider."""
    return callable(getattr(obj, 'mgr_return_value', None))


class ObjectFormatAdapter:
    """A format adapater for a single object.
    Given an input object, this type will adapt the object, or a simplified
    representation of the object, to either JSON or YAML when the format_json or
    format_yaml methods are used.

    If the compatible flag is true and the object provided to the adapter has
    methods such as `to_json` and/or `to_yaml` these methods will be called in
    order to get a JSON/YAML compatible simplified representation of the
    object.

    If the above case is not satisfied and the object provided to the adapter
    has a method `to_simplified`, this method will be called to acquire a
    simplified representation of the object.

    If none of the above cases is true, the object itself will be used for
    serialization. If the object can not be safely serialized an exception will
    be raised.

    NOTE: Some code may use methods named like `to_json` to return a JSON
    string. If that is the case, you should not use that method with the
    ObjectFormatAdapter. Do not set compatible=True for objects of this type.
    """

    def __init__(
        self,
        obj: Any,
        json_indent: Optional[int] = DEFAULT_JSON_INDENT,
        compatible: bool = False,
    ) -> None:
        self.obj = obj
        self._compatible = compatible
        self.json_indent = json_indent

    def _fetch_json_data(self) -> Any:
        # if the data object provides a specific simplified representation for
        # JSON (and compatible mode is enabled) get the data via that method
        if self._compatible and _is_json_data_provider(self.obj):
            return self.obj.to_json()
        # otherwise we use our specific method `to_simplified` if it exists
        if _is_simple_data_provider(self.obj):
            return self.obj.to_simplified()
        # and fall back to the "raw" object
        return self.obj

    def format_json(self) -> str:
        """Return a JSON formatted string representing the input object."""
        return json.dumps(
            self._fetch_json_data(), indent=self.json_indent, sort_keys=True
        )

    def _fetch_yaml_data(self) -> Any:
        if self._compatible and _is_yaml_data_provider(self.obj):
            return self.obj.to_yaml()
        # nothing specific to YAML was found. use the simplified representation
        # for JSON, as all valid JSON is valid YAML.
        return self._fetch_json_data()

    def format_yaml(self) -> str:
        """Return a YAML formatted string representing the input object."""
        return yaml.safe_dump(self._fetch_yaml_data())

    format_json_pretty = format_json

    def valid_formats(self) -> Iterable[str]:
        """Return valid format names."""
        return set(str(v) for v in Format.__members__)


class ReturnValueAdapter:
    """A return-value adapter for an object.
    Given an input object, this type will attempt to get a mgr return value
    from the object if provides a `mgr_return_value` function.
    If not it returns a default return value, typically 0.
    """

    def __init__(
        self,
        obj: Any,
        default: int = 0,
    ) -> None:
        self.obj = obj
        self.default_return_value = default

    def mgr_return_value(self) -> int:
        if _is_return_value_provider(self.obj):
            return int(self.obj.mgr_return_value())
        return self.default_return_value


class ErrorResponseBase(Exception):
    """An exception that can directly be converted to a mgr reponse."""

    def format_response(self) -> Tuple[int, str, str]:
        raise NotImplementedError()


class UnknownFormat(ErrorResponseBase):
    """Raised if the format name is unexpected.
    This can help distinguish typos from formats that are known but
    not implemented.
    """

    def __init__(self, format_name: str) -> None:
        self.format_name = format_name

    def format_response(self) -> Tuple[int, str, str]:
        return -errno.EINVAL, "", f"Unknown format name: {self.format_name}"


class UnsupportedFormat(ErrorResponseBase):
    """Raised if the format name does not correspond to any valid
    conversion functions.
    """

    def __init__(self, format_name: str) -> None:
        self.format_name = format_name

    def format_response(self) -> Tuple[int, str, str]:
        return -errno.EINVAL, "", f"Unsupported format: {self.format_name}"


class ErrorResponse(ErrorResponseBase):
    """General exception convertible to a mgr response."""

    E = TypeVar("E", bound="ErrorResponse")

    def __init__(self, status: str, return_value: Optional[int] = None) -> None:
        self.return_value = (
            return_value if return_value is not None else -errno.EINVAL
        )
        self.status = status

    def format_response(self) -> Tuple[int, str, str]:
        return (self.return_value, "", self.status)

    def mgr_return_value(self) -> int:
        return self.return_value

    @property
    def errno(self) -> int:
        rv = self.return_value
        return -rv if rv < 0 else rv

    def __repr__(self) -> str:
        return f"ErrorResponse({self.status!r}, {self.return_value!r})"

    @classmethod
    def wrap(
        cls: Type[E], exc: Exception, return_value: Optional[int] = None
    ) -> ErrorResponseBase:
        if isinstance(exc, ErrorResponseBase):
            return exc
        if return_value is None:
            try:
                return_value = int(getattr(exc, "errno"))
                if return_value > 0:
                    return_value = -return_value
            except (AttributeError, ValueError):
                pass
        err = cls(str(exc), return_value=return_value)
        setattr(err, "__cause__", exc)
        return err


ObjectResponseFuncType = Union[
    Callable[..., Dict[Any, Any]],
    Callable[..., List[Any]],
    Callable[..., SimpleDataProvider],
    Callable[..., JSONDataProvider],
    Callable[..., YAMLDataProvider],
    Callable[..., ReturnValueProvider],
]


def _get_requested_format(f: ObjectResponseFuncType, kw: Dict[str, Any]) -> str:
    # todo: leave 'format' in kw dict iff its part of f's signature
    return kw.pop("format", None)


class Responder:
    """A decorator type intended to assist in converting Python return types
    into valid responses for the Ceph MGR.

    A function that returns a Python object will have the object converted into
    a return value and formatted response body, based on the `format` argument
    passed to the mgr. When used from the ceph cli tool the `--format=[name]`
    argument is mapped to a `format` keyword argument. The decorated function
    may provide a `format` argument (type str). If the decorated function does
    not provide a `format` argument itself, the Responder decorator will
    implicitly add one to the MGR's "CLI arguments" handling stack.

    The Responder object is callable and is expected to be used as a decorator.
    """

    def __init__(
        self, formatter: Optional[Callable[..., CommonFormatter]] = None
    ) -> None:
        self.formatter = formatter
        self.default_format = "json"

    def _formatter(self, obj: Any) -> CommonFormatter:
        """Return the formatter/format-adapter for the object."""
        if self.formatter is not None:
            return self.formatter(obj)
        return ObjectFormatAdapter(obj)

    def _retval_provider(self, obj: Any) -> ReturnValueProvider:
        """Return a ReturnValueProvider for the given object."""
        return ReturnValueAdapter(obj)

    def _get_format_func(
        self, obj: Any, format_req: Optional[str] = None
    ) -> Callable:
        formatter = self._formatter(obj)
        if format_req is None:
            format_req = self.default_format
        if format_req not in formatter.valid_formats():
            raise UnknownFormat(format_req)
        req = str(format_req).replace("-", "_")
        ffunc = getattr(formatter, f"format_{req}", None)
        if ffunc is None:
            raise UnsupportedFormat(format_req)
        return ffunc

    def _dry_run(self, format_req: Optional[str] = None) -> None:
        """Raise an exception if the format_req is not supported."""
        # call with an empty dict to see if format_req is valid and supported
        self._get_format_func({}, format_req)

    def _formatted(self, obj: Any, format_req: Optional[str] = None) -> str:
        """Return the object formatted/serialized."""
        ffunc = self._get_format_func(obj, format_req)
        return ffunc()

    def _return_value(self, obj: Any) -> int:
        """Return a mgr return-value for the given object (usually zero)."""
        return self._retval_provider(obj).mgr_return_value()

    def __call__(self, f: ObjectResponseFuncType) -> HandlerFuncType:
        """Wrap a python function so that the original function's return value
        becomes the source for an automatically formatted mgr response.
        """

        @wraps(f)
        def _format_response(*args: Any, **kwargs: Any) -> Tuple[int, str, str]:
            format_req = _get_requested_format(f, kwargs)
            try:
                self._dry_run(format_req)
                robj = f(*args, **kwargs)
                body = self._formatted(robj, format_req)
                retval = self._return_value(robj)
            except ErrorResponseBase as e:
                return e.format_response()
            return retval, body, ""

        # set the extra args on our wrapper function. this will be consumed by
        # the CLICommand decorator and added to the set of optional arguments
        # on the ceph cli/api
        setattr(_format_response, "extra_args", {"format": str})
        return _format_response


class ErrorResponseHandler:
    """ErrorResponseHandler is a very simple decorator that handles functions that
    raise exceptions inheriting from ErrorResponseBase. If such an exception
    is raised that exception can and will be converted to a mgr response tuple.
    This is similar to Responder but error handling is all this decorator does.
    """

    def __call__(self, f: Callable[..., Tuple[int, str, str]]) -> HandlerFuncType:
        """Wrap a python function so that if the function raises an exception inheriting
        ErrorResponderBase the error is correctly converted to a mgr response.
        """

        @wraps(f)
        def _format_response(*args: Any, **kwargs: Any) -> Tuple[int, str, str]:
            try:
                retval, body, sts = f(*args, **kwargs)
            except ErrorResponseBase as e:
                return e.format_response()
            return retval, body, sts

        return _format_response


class ConstantResponderBase:
    """The constant responder base assumes that a wrapped function should not
    be passing data back to the manager. It only responds with the default
    (constant) values provided. The process_response function allows a subclass
    to handle/log/validate any values that were returned from the wrapped
    function.

    This class can be used a building block for special decorators that
    do not normally emit response data.
    """

    def mgr_return_value(self) -> int:
        return 0

    def mgr_body_value(self) -> str:
        return ""

    def mgr_status_value(self) -> str:
        return ""

    def process_response(self, result: Any) -> None:
        return None

    def __call__(self, f: Callable) -> HandlerFuncType:
        """Wrap a python function so that if the function raises an exception
        inheriting ErrorResponderBase the error is correctly converted to a mgr
        response. Otherwise, it returns a default set of constant values.
        """

        @wraps(f)
        def _format_response(*args: Any, **kwargs: Any) -> Tuple[int, str, str]:
            try:
                self.process_response(f(*args, **kwargs))
            except ErrorResponseBase as e:
                return e.format_response()
            return self.mgr_return_value(), self.mgr_body_value(), self.mgr_status_value()
        return _format_response


class EmptyResponder(ConstantResponderBase):
    """Always respond with an empty (string) body. Checks that the wrapped function
    returned None in order to ensure it is not being used on functions that
    return data objects.
    """

    def process_response(self, result: Any) -> None:
        if result is not None:
            raise ValueError("EmptyResponder expects None from wrapped functions")