From b9a66e14d9dee6a00c82f6e6f772a023050e8da1 Mon Sep 17 00:00:00 2001 From: John Mulligan Date: Tue, 30 Jan 2024 14:39:16 -0500 Subject: [PATCH] pybind/mgr/smb: add resourcelib.py an internal resource mgmt lib While I like the workflow that `ceph orch apply` provides I find the code a little too "loose". Create a new minimalistic un/re-structuring library that partly inspired by my work with Go, cephadm, and a little from pydantic. But without adding any dependencies beyond python's dataclasses. Signed-off-by: John Mulligan --- src/pybind/mgr/smb/resourcelib.py | 641 ++++++++++++++++++++++++++++++ 1 file changed, 641 insertions(+) create mode 100644 src/pybind/mgr/smb/resourcelib.py diff --git a/src/pybind/mgr/smb/resourcelib.py b/src/pybind/mgr/smb/resourcelib.py new file mode 100644 index 0000000000000..6d3a7ff63c762 --- /dev/null +++ b/src/pybind/mgr/smb/resourcelib.py @@ -0,0 +1,641 @@ +"""resourcelib - a semi-automated library for structured objects + +This library aims to be both compatible with the mgr's object_format library +and be a base for a cleaner looking & easier working way to accomplish things +like the `ceph nfs export apply` command and the `ceph orch apply` command +but for smb. This only defines the tools to do semi-automated structuring/unstructuring +- serialization into JSON/YAML can be handled at higher levels but without +needing to get fancy in those libraries. + +Quick Example: +>>> from . import resourcelib +>>> @resourcelib.component() +... class Rect: +... x_pos: int +... y_pos: int +... width: int +... height: int +>>> @resourcelib.resource('art') +... class Art: +... name: str +... rectangles: List[Rect] +>>> rdata = { +... 'resource_type': 'art', +... 'name': 'My Paintings', +... 'rectangles': [ +... {'x_pos': 5, 'y_pos': 0, 'width': 12, 'height': 9}, +... {'x_pos': 20, 'y_pos': 35, 'width': 10, 'height': 40}, +... ], +... } +>>> a = resourcelib.load(rdata) +>>> isinstance(a[0], Art) +True + +That is, a dictionary containing a `resource_type` field and other data +fields can be easily converted into a more structured python object +based on dataclasses and a few decorators. This object can also be +converted back into a less-structured dict using the automatically +created `to_simplified` method. + +One can also work directly with the Resource objects that are created +for each class. First, the special method `_customize_resource` with +the `customize` decorator can be used to alter the Resource immediately +after it is created. Alternatively, the property `_resource_config` is +added to every resourcelib component or resource. + +Example: +>>> @resourcelib.resource('widget') +... class Widget: +... name: str +... description: str = '' +... labels: Optional[List[str]] = None +... @resourcelib.customize +... def _customize_resource(rc: resourcelib.Resource) -> resourcelib.Resource: +... # set description field quiet property to true. this will hide +... # empty strings from the unstructured dict +... rc.description.quiet = True +... return rc +>>> def do_something(): +... "Do something silly with the widget class resource." +... w = Widget('default', 'A generic widget') +... return Widget._resource_config.object_to_simplified(w) +>>> do_something() +{'resource_type': 'widget', 'name': 'default', 'description': 'A generic widget'} + +Technically, in the _customize_resource method, you can manipulate the Resource +object deeply or even return a truly customized Resource subclass object +if you want to! + +Multiple classes can share a single `resource_type` name if, and only if, +all resources are configured to have a condition. A condition is a function +that takes the raw dict and returns true/false indicating that the resource +and the dataclass it handles are appropriate for the data in question. +One can set the condition function using the `Resource.on_condition` method. +This can be invoked via the `_customize_resource` method. + +The library doesn't check if the conditions are comprehensive. If you have +two classes mapped to resource_type "x" and neither condition returns true +the library will simply raise an exception. +""" +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Dict, + Hashable, + List, + Optional, + Tuple, +) + +import dataclasses +import logging +import sys +from itertools import chain + +from .proto import Self, Simplified + +log = logging.getLogger(__name__) + +_RESOURCE_TYPE = 'resource_type' +_RESOURCES = 'resources' +_DEBUG = False + + +if _DEBUG: + + def _xt(f: Callable) -> Callable: + def _func(*args: Any, **kwargs: Any) -> Any: + log.debug(f'\nCALL({f}): {args!r}, {kwargs!r}') + try: + result = f(*args, **kwargs) + log.debug(f'\n RET({f}): result={result!r}') + return result + except Exception as err: + log.debug(f'\n EXC({f}): {err!r}') + raise + + return _func + +else: + + def _xt(f: Callable) -> Callable: + return f + + +if sys.version_info >= (3, 11): + from typing import dataclass_transform +elif TYPE_CHECKING: + from typing_extensions import dataclass_transform +else: + # no-op decorator, placeholder for dataclass_transform + def dataclass_transform(*args: Any, **kwargs: Any) -> Callable: + def _phony(f: Callable) -> Callable: + return f + + return _phony + + +# ---- Error Types ---- + + +class ResourceTypeError(ValueError): + """Generic error working with smb resource types.""" + + pass + + +class MissingResourceTypeError(KeyError, ResourceTypeError): + """Exception raised when converting from unstructured data and a required + `resource_type` field is missing. + """ + + def __init__(self, data: Simplified) -> None: + self.data = data + + def __str__(self) -> str: + return 'source data is missing a resource_type field' + + +class InvalidResourceTypeError(ResourceTypeError): + """Exception raised when an object can not be converted from unstructured + data. + """ + + def __init__(self, *, expected: Any = None, actual: Any = None) -> None: + self.expected = expected + self.actual = actual + + def __str__(self) -> str: + msg = f'invalid resource type value: {self.actual!r}' + if self.expected: + msg += f'; expected: {self.expected!r}' + return msg + + +class MissingRequiredFieldError(KeyError, ResourceTypeError): + """Exception raised when an object can not be converted from unstructured + data due to a missing required field. + """ + + def __init__(self, key: str) -> None: + self.key = key + + def __str__(self) -> str: + return f'data object missing required field: {self.key}' + + +# ---- Internal Resource Types ---- + +# Sentinel object for unset/missing value conditions. +_unset = object() + + +def _unwrap_type(atype: Any) -> Tuple[Tuple, bool]: + """Given an Optional[T] type return (T, True). Given a non-optional type + return the (T, false). + """ + args = _get_args(atype) + uargs: Tuple[Any, ...] = tuple(a for a in args if a is not type(None)) + return uargs, bool(args and len(args) != len(uargs)) + + +def _get_args(atype: Any) -> Tuple: + """Given a type object return the types args. + Example: List[T] -> (T,), Union[A, B] -> (A, B). + """ + return getattr(atype, '__args__', tuple()) + + +def _get_origin(atype: Any) -> Any: + """Given a type such as List[T] return the outer type (List).""" + return getattr(atype, '__origin__', None) + + +@dataclasses.dataclass +class Field: + """Metadata about a field (member) of a resource type class.""" + + name: str + field_type: Any + default: Any = _unset + quiet: bool = False # customization value + keep_none: bool = False # customization value + + def optional(self) -> bool: + """Return true if the type of the field is Optional.""" + _, optional = _unwrap_type(self.field_type) + return optional + + def inner_type(self) -> Any: + """For a field with an optional type (Optional[T]) return the + nested type (T). Otherwise return the current field type. + """ + args, optional = _unwrap_type(self.field_type) + if optional: + assert len(args) == 1 + return args[0] + return self.field_type + + @_xt + def takes(self, dest_type: Any) -> bool: + """Returns true if the field's type is composed of a matching container + type. It supports field types of the form `List[T]`, `Optional[List[T]]`, + `Dict[T, V]`, `Optional[Dict[T, V]]`. + + For example a field with type `List[str]` returns true for + `f.takes(list)`. Similarly, `Optional[List[str]]` returns true for + `f.takes(list)`. A field `Dict[int, int]` returns true for `f.takes(dict)`, + and so on. + + Ideally newer-style types like `list[T] | None` supported by later + python versions should also work but this has not been tested. + """ + otype = _get_origin(self.inner_type()) + if otype is None: + return False + + if dest_type is list: + dest_type = (list, List) + elif dest_type is dict: + dest_type = (dict, Dict) + elif not isinstance(dest_type, tuple): + dest_type = (dest_type,) + return otype in dest_type + + def list_element_type(self) -> Any: + """Assuming the field type is a list (List[T]) return the type + of the list's elements (T). + """ + args, optional = _unwrap_type(self.inner_type()) + assert not optional + assert len(args) == 1 + return args[0] + + def dict_element_types(self) -> Tuple[Any, Any]: + """Assuming the field type is a dict (List[KT, VT]) return the types + of the dict's keys & values (KT, VT). + """ + args, optional = _unwrap_type(self.inner_type()) + assert not optional + assert len(args) == 2 + return args[0], args[1] + + @classmethod + def create(cls, fld: dataclasses.Field) -> Self: + """Contructor converting from a dataclasses.Field.""" + default = _unset + if fld.default is not dataclasses.MISSING: + default = fld.default + return cls( + name=fld.name, + field_type=fld.type, + default=default, + ) + + +class Resource: + """Type converter that maintains metadata about a structured python object + and can be used to convert to / from the structured object and unstructured + (JSON/YAML-safe dict) data. + """ + + def __init__(self, cls: Any) -> None: + self.resource_cls = cls + self.fields: Dict[str, Field] = {} + self._on_condition: Optional[Callable[..., bool]] = None + + for fld in dataclasses.fields(self.resource_cls): + self.fields[fld.name] = Field.create(fld) + + @property + def conditional(self) -> bool: + """Return true if the resource is selectable based on a condition.""" + return self._on_condition is not None + + def on_condition(self, cond: Callable[..., bool]) -> None: + """Set a condition function.""" + self._on_condition = cond + + def type_name(self) -> str: + """Return the name of the type managed by this resource.""" + return self.resource_cls.__name__ + + def __getattr__(self, name: str) -> Field: + """Return a field metadata object for the type managed by this resource.""" + return self.fields[name] + + @_xt + def object_from_simplified(self, data: Simplified) -> Any: + """Given a dict-based unstructured data object return the structured + object-based equivalent. + """ + kw = {} + for fld in self.fields.values(): + value = self._object_field_from_simplified(fld, data) + if value is not _unset: + kw[fld.name] = value + obj = self.resource_cls(**kw) + validate = getattr(obj, 'validate', None) + if validate: + validate() + return obj + + @_xt + def _object_field_from_simplified( + self, fld: Field, data: Simplified + ) -> Any: + if fld.name not in data and fld.default is not _unset: + return _unset + elif fld.name not in data: + raise MissingRequiredFieldError(fld.name) + + value = data[fld.name] + if value is None and fld.optional(): + return None + inner_type = fld.inner_type() + + _rconfig = getattr(inner_type, '_resource_config', None) + if _rconfig: + return _rconfig.object_from_simplified(value) + _fs = getattr(inner_type, 'from_simplified', None) + if _fs: + return _fs(value) + + if fld.takes(list): + subtype = fld.list_element_type() + return [ + self._object_sub_from_simplified(subtype, v) for v in value + ] + if fld.takes(dict): + ktype, vtype = fld.dict_element_types() + # keys must be simple types right now so we just + # cast it directly + return { + ktype(k): self._object_sub_from_simplified(vtype, v) + for k, v in value.items() + } + + return inner_type(value) + + @_xt + def _object_sub_from_simplified( + self, subtype: Any, data: Simplified + ) -> Any: + _rconfig = getattr(subtype, '_resource_config', None) + if _rconfig: + return _rconfig.object_from_simplified(data) + if _get_origin(subtype) in (list, List): + return list(data) + if _get_origin(subtype) in (dict, Dict): + return dict(data) + return subtype(data) + + @_xt + def object_to_simplified(self, obj: Any) -> Simplified: + """Given a python object tagged as a resource type return the + unstructured data equivalent. + """ + result: Simplified = {} + rt = getattr(obj, _RESOURCE_TYPE, None) + if rt is not None: + result[_RESOURCE_TYPE] = rt + for fld in self.fields.values(): + self._object_field_to_simplified(obj, fld, result) + return result + + @_xt + def _object_field_to_simplified( + self, obj: Any, fld: Field, data: Simplified + ) -> None: + value = getattr(obj, fld.name) + assert fld.optional() or value is not None + if (value is None and not fld.keep_none) or ( + fld.quiet and value is not None and not value + ): + return + + _rconfig = getattr(fld.inner_type(), '_resource_config', None) + if _rconfig: + data[fld.name] = _rconfig.object_to_simplified(value) + return + _ts = getattr(fld.inner_type(), 'to_simplified', None) + if _ts: + data[fld.name] = _ts(value) + return + + if isinstance(value, list): + assert fld.takes(list) + subtype = fld.list_element_type() + data[fld.name] = [ + self._object_sub_to_simplified(subtype, v) for v in value + ] + return + if isinstance(value, dict): + assert fld.takes(dict) + ktype, vtype = fld.dict_element_types() + data[fld.name] = { + ktype(k): self._object_sub_to_simplified(vtype, v) + for k, v in value.items() + } + return + + if isinstance(value, str): + data[fld.name] = str(value) + return + if isinstance(value, (int, float)): + data[fld.name] = value + return + raise ResourceTypeError(f'unexpected type for field {fld.name}') + + @_xt + def _object_sub_to_simplified(self, subtype: Any, value: Any) -> Any: + _rconfig = getattr(subtype, '_resource_config', None) + if _rconfig: + return _rconfig.object_to_simplified(value) + if isinstance(value, dict): + return value + if isinstance(value, list): + return value + if isinstance(value, str): + return str(value) + if isinstance(value, (int, float)): + return value + raise ResourceTypeError(f'unexpected type: {type(value)}') + + def __repr__(self) -> str: + return ( + f'{self.__class__.__name__}<' + f'{self.type_name()}, conditional={self.conditional}>' + ) + + @classmethod + def create(cls, resource_cls: Any) -> Self: + """Constructor that creates a Resource for a given python class.""" + resource = cls(resource_cls) + _customize = getattr(resource_cls, '_customize_resource', None) + if _customize is not None: + resource = _customize(resource) + return resource + + +class Registry: + """Registry to track resource objects.""" + + def __init__(self) -> None: + self.resources: Dict[str, List[Resource]] = {} + self.types: Dict[Hashable, Resource] = {} + + def enable(self, cls: Any) -> Resource: + """Given a python class create and record resource for it. Return the + new resource. + """ + resource = Resource.create(cls) + self.types[cls] = resource + return resource + + def track(self, key: str, resource: Resource) -> None: + """Given a resource-type-name and a resource object, save these items + into the registry. A key may be repeated creating a "discriminating + union" so long as the resource type provides a condition function to + determine what resource to use. + """ + if key not in self.resources: + self.resources[key] = [] + self.resources[key].append(resource) + if len(self.resources[key]) > 1: + if any(not c.conditional for c in self.resources[key]): + raise ResourceTypeError( + 'multiplexed resources must be conditional' + ) + + def select(self, data: Simplified) -> Resource: + """Given an unstructured data object use the resource_type field and + any condition functions to return the resource object that matches. + """ + try: + rt = data[_RESOURCE_TYPE] + except KeyError: + raise MissingResourceTypeError(data) + try: + matches = self.resources[rt] + except KeyError: + raise InvalidResourceTypeError(actual=rt) + if len(matches) == 1: + return matches[0] + for rc in matches: + if not rc._on_condition: + raise ResourceTypeError('resource not conditional') + if rc._on_condition(data): + return rc + raise ResourceTypeError('no resource type conditions met') + + +class _ResourceType: + """A read-only property of a class acting as a resource type. When accessed, + returns the name of the resource type. + """ + + def __init__(self, name: str) -> None: + self.name = name + + def __get__(self, obj: Any, objtype: Any = None) -> str: + return self.name + + +class _ResourceConfig: + """A read-only property of a class acting as a resource type that, when + accessed, returns the resource object. + """ + + def __init__(self, resource: Resource) -> None: + self.resource = resource + + def __get__(self, obj: Any, objtype: Any = None) -> Resource: + return self.resource + + +# _REGISTRY is our internal registry object +_REGISTRY = Registry() + + +# ---- decorators ---- + + +def _to_simplified(obj: Any) -> Simplified: + rc = getattr(obj, '_resource_config') + return rc.object_to_simplified(obj) + + +def _make_resource(cls: Any, resource_name: str = '') -> Any: + cls = dataclasses.dataclass(cls) + rconfig = _REGISTRY.enable(cls) + cls._resource_config = _ResourceConfig(rconfig) + if resource_name: + cls.resource_type = _ResourceType(resource_name) + _REGISTRY.track(resource_name, rconfig) + if getattr(cls, 'to_simplified', None) is None: + cls.to_simplified = _to_simplified + return cls + + +@dataclass_transform() +def resource(resource_name: str) -> Any: + """Class-decorator that establishes a new dataclass/resource type. + These types can be converted from raw data using the `load` function. + """ + assert resource_name + + def _decorator(cls: Any) -> Any: + return _make_resource(cls, resource_name) + + return _decorator + + +@dataclass_transform() +def component() -> Any: + """Class-decorator that establishes a new dataclass/resource type without a + top-level name. These types may be manually converted from raw or embedded + within a resource type. + """ + + def _decorator(cls: Any) -> Any: + return _make_resource(cls) + + return _decorator + + +# TODO: make customize validate the function and not just a wrapper over +# staticmethod +customize = staticmethod + + +# --- resource load function ---- + + +@_xt +def load(data: Simplified) -> List[Any]: + """Given a simple unstructured data object (python dict/list) containing + only other simple data types, use the `resource_type` metadata to + convert the input to a list of structured and typed objects. + + The input may be: + * a single item: {"resource_type": "foo", ...} which will return + a single valued list + * a python list containing dicts like the above + * a dict containing the key "resources" that maps to a list containing dicts + like the 1st item + """ + # Given a bare list/iterator. Assume it contains loadable objects. + if not isinstance(data, dict): + return list(chain.from_iterable(load(v) for v in data)) + # Given a "list object" + if _RESOURCE_TYPE not in data and _RESOURCES in data: + rl = data[_RESOURCES] + if not isinstance(rl, list): + raise TypeError('expected resources list') + return list(chain.from_iterable(load(v) for v in rl)) + # anything else must be a "self describing" object with a resource_type + # value + resource = _REGISTRY.select(data) + return [resource.object_from_simplified(data)] -- 2.39.5