Source code for ulkb.object

# Copyright (C) 2023 IBM Corp.
# SPDX-License-Identifier: Apache-2.0

from abc import ABCMeta, abstractmethod
from collections.abc import Sequence

from . import error, util

__all__ = [
    'Object',
]


class ObjectMeta(ABCMeta):

    _object_class = None

    def __new__(mcls, name, bases, namespace, **kwargs):
        cls = super().__new__(mcls, name, bases, namespace, **kwargs)
        mcls._init(cls, name, bases, namespace, **kwargs)
        return cls

    @classmethod
    def _init(mcls, cls, name, bases, namespace, **kwargs):
        top = mcls._object_class or cls
        setattr(top, name, cls)
        cls.name = util.camel2snake(name)
        mcls._init_is_(top, cls)
        mcls._init_test_(top, cls)
        mcls._init_check_(top, cls)
        mcls._init_unfold_(top, cls)
        mcls._init_unpack_(top, cls)
        mcls._init_preprocess_arg_(top, cls)
        mcls._init_cached_(top, cls)
        if top == cls:          # execute only once, for top
            mcls._init_converters(top)
            mcls._init_parsers(top)
            mcls._init_serializers(top)
        return cls

    @classmethod
    def _init_is_(mcls, top, cls):
        setattr(top, 'is_' + cls.name, lambda x: cls.test(x))

    @classmethod
    def _init_test_(mcls, top, cls):
        def f_test(arg):
            return cls.test(arg)
        f_test.__doc__ = f"""\
        Tests whether object is an instance of :class:`{cls.__name__}`.

        Returns:
           ``True`` if successful; ``False`` otherwise.
        """
        setattr(top, 'test_' + cls.name, f_test)

    @classmethod
    def _init_check_(mcls, top, cls):
        def mk_check_(s):
            def check_(arg, func_name=s, arg_name=None, arg_position=None):
                return cls.check(arg, func_name, arg_name, arg_position)
            return check_
        s = 'check_' + cls.name
        f_check = mk_check_(s)
        f_check.__doc__ = f"""\
        Checks whether object is an instance of :class:`{cls.__name__}`.

        Parameters:
           func_name: Function name.
           arg_name: Argument name.
           arg_position: Argument position.

        Returns:
           :class:`{cls.__name__}`.

        Raises:
           TypeError: Object is not an instance of :class:`{cls.__name__}`.
        """
        setattr(top, s, f_check)

    @classmethod
    def _init_unfold_(mcls, top, cls):
        def f_unfold(arg):
            return cls.unfold(arg)
        f_unfold.__doc__ = f"""\
        Unfolds arguments of :class:`{cls.__name__}`.

        Returns:
           :class:`{cls.__name__}`'s arguments unfolded.

        Raises:
           TypeError: Object is not an instance of :class:`{cls.__name__}`.
        """

        def f_unfold_unsafe(arg):
            return cls.unfold_unsafe(arg)
        f_unfold_unsafe.__doc__ = f"""\
        Unfolds arguments of :class:`{cls.__name__}` (unsafe version).

        Returns:
           :class:`{cls.__name__}`'s arguments unfolded if object is an
           instance of :class:`{cls.__name__}`; ``None`` otherwise.
        """

        s = 'unfold_' + cls.name
        setattr(top, s, f_unfold)
        setattr(top, s + '_unsafe', f_unfold_unsafe)
        setattr(top, '_' + s, lambda x: cls._unfold(x))

    @classmethod
    def _init_unpack_(mcls, top, cls):
        def f_unpack(arg):
            return cls.unpack(arg)
        f_unpack.__doc__ = f"""\
        Unpacks arguments of :class:`{cls.__name__}`.

        Returns:
           :class:`{cls.__name__}`'s arguments unpacked.

        Raises:
           TypeError: Object is not an instance of :class:`{cls.__name__}`.
        """

        def f_unpack_unsafe(arg):
            return cls.unpack_unsafe(arg)
        f_unpack_unsafe.__doc__ = f"""\
        Unpacks arguments of :class:`{cls.__name__}` (unsafe version).

        Returns:
           :class:`{cls.__name__}`'s arguments unpacked if object is an
           instance of :class:`{cls.__name__}`; ``None`` otherwise.
        """

        s = 'unpack_' + cls.name
        setattr(top, s, f_unpack)
        setattr(top, s + '_unsafe', f_unpack_unsafe)
        setattr(top, '_' + s, lambda x: cls._unpack(x))

    @classmethod
    def _init_preprocess_arg_(mcls, top, cls):
        s = '_preprocess_arg_' + cls.name
        if hasattr(cls, s):
            setattr(top, s, staticmethod(getattr(cls, s)))
        else:
            def mk__preprocess_arg_(c):
                def _preprocess_arg_(self, arg, i):
                    return c.check(arg, self.__class__.__name__, None, i)
                return _preprocess_arg_
            setattr(top, s, staticmethod(mk__preprocess_arg_(cls)))

    @classmethod
    def _init_cached_(mcls, top, cls):
        setattr(cls, '_cached', list(filter(
                lambda x: x.startswith('_cached_'), cls.__dict__)))
        if cls._cached:
            def mk__init_cached(cached):
                def _init_cached(self):
                    super(cls, self)._init_cached()
                    for attr in cached:
                        setattr(self, attr, None)
                return _init_cached
            setattr(cls, '_init_cached', mk__init_cached(cls._cached))

            def mk_get_cached(x, suffix, _get):
                def get_cached(self):
                    if getattr(self, x) is None:
                        setattr(self, x, _get(self))
                    return getattr(self, x)
                return get_cached
            for attr in cls._cached:
                suffix = attr[8:]
                f_build = getattr(cls, '_build_' + suffix + '_cache')
                f_get = mk_get_cached(attr, suffix, f_build)
                f_get.__doc__ = f_build.__doc__
                setattr(cls, 'get_' + suffix, f_get)
                f_get_prop = mk_get_cached(attr, suffix, f_build)
                doc = f_build.__doc__
                if doc:
                    doc = doc[9].capitalize() + doc[10:]
                f_get_prop.__doc__ = doc
                setattr(cls, suffix, property(f_get_prop))

    @classmethod
    def _init_converters(mcls, top):
        from .converter import Converter
        for format, v in Converter.converters.items():
            def mk_from_(fmt):
                def from_(obj_cls, value, **kwargs):
                    return obj_cls.convert_from(
                        value, format=fmt, **kwargs)
                return from_
            format_long = v.get('format_long', format)
            f_from = mk_from_(format)
            f_from.__doc__ = f"""\
            Converts {format_long} `value` to object.

            Parameters:
               value: {format_long} value.
               kwargs: Options to converter.

            Returns:
               The resulting :class:`Object`.
            """
            setattr(top, 'from_' + format, classmethod(f_from))

            def mk_to_(fmt):
                def to_(self, **kwargs):
                    return self.convert_to(format=fmt, **kwargs)
                return to_
            f_to = mk_to_(format)
            f_to.__doc__ = f"""\
            Converts object to {format_long} value.

            Parameters:
               kwargs: Options to converter.

            Returns:
               The resulting {format_long} value.
            """
            setattr(top, 'to_' + format, f_to)

    @classmethod
    def _init_parsers(mcls, top):
        from .parser import Parser
        for format, v in Parser.parsers.items():
            def mk_from_(fmt):
                def from_(
                        obj_cls, from_=None, path=None, encoding=None,
                        **kwargs):
                    return obj_cls.parse(
                        from_=from_, path=path,
                        encoding=encoding, format=fmt, **kwargs)
                return from_
            format_long = v.get('format_long', format)
            f_from = mk_from_(format)
            f_from.__doc__ = f"""\
            Parses {format_long} stream into object.

            Parameters:
               from_: Source string or stream object.
               path: Source file path.
               encoding: Encoding.
               kwargs: Options to parser.

            Returns:
               The resulting :class:`Object`.
            """
            setattr(top, 'from_' + format, classmethod(f_from))

    @classmethod
    def _init_serializers(mcls, top):
        from .serializer import Serializer
        for format, v in Serializer.serializers.items():
            def mk_to_(fmt):
                def to_(obj, to=None, encoding=None, **kwargs):
                    return obj.serialize(
                        to=to, encoding=encoding, format=fmt, **kwargs)
                return to_
            format_long = v.get('format_long', format)
            f_to = mk_to_(format)
            f_to.__doc__ = f"""\
            Serializes object to {format_long} stream.

            Parameters:
               to: Target file path or stream object.
               encoding: Encoding.
               kwargs: Options to serializer.

            Returns:
               The resulting string if `to` is ``None``.  Otherwise, write
               the resulting string to `to` and returns ``None``.
            """
            setattr(top, 'to_' + format, f_to)


[docs] @util.total_ordering class Object(Sequence, metaclass=ObjectMeta): """Abstract base class for syntactical objects. An :class:`Object` consists of a tuple of arguments :attr:`args` together with a dictionary of annotations :attr:`annotations`. Parameters: args: Arguments kwargs: Annotations. Returns: :class:`Object`. """ @classmethod def _thy(cls, thy=None): return thy or cls.Theory.top @classmethod def _dup(cls, *args, **kwargs): return cls(*args, **kwargs) @classmethod def _is_generated_id(cls, id): return id.startswith(cls._thy().settings.generated_id_prefix)
[docs] @classmethod def test(cls, arg): """Tests whether `arg` is an instance of this class. Parameters: arg: Value. Returns: ``True`` if successful; ``False`` otherwise. """ return isinstance(arg, cls)
[docs] @classmethod def check(cls, arg, func_name=None, arg_name=None, arg_position=None): """Checks whether `arg` is an instance of this class. Parameters: arg: Value. func_name: Function name. arg_name: Argument name. arg_position: Argument position. Returns: `arg`. Raises: TypeError: `arg` is not an instance of this class. """ return error.check_arg_class_test( arg, cls, cls.test(arg), func_name or 'check', arg_name, arg_position)
[docs] @classmethod def unfold(cls, arg): """Unfolds `arg`'s arguments. Parameters: arg: Value. Returns: `arg`'s arguments unfolded. Raises: TypeError: `arg` is not an instance of this class. """ return cls._unfold(cls.check(arg, 'unfold'))
[docs] @classmethod def unfold_unsafe(cls, arg): """Unfolds `arg`'s arguments (unsafe version). Parameters: arg: Value. Returns: `arg`'s arguments unfolded if `arg` is an instance of this class; ``None`` otherwise. """ return cls._unfold(arg) if cls.test(arg) else None
@classmethod def _unfold(cls, arg): return cls._unpack(arg)
[docs] @classmethod def unpack(cls, arg): """Unpacks `arg`'s arguments. Parameters: arg: Value. Returns: `arg`'s arguments unpacked. Raises: TypeError: `arg` is not an instance of this class. """ return cls._unpack(cls.check(arg, 'unpack'))
[docs] @classmethod def unpack_unsafe(cls, arg): """Unpacks `arg`'s arguments (unsafe version). Parameters: arg: Value. Returns: `arg`'s arguments unpacked if `arg` is an instance of this class; ``None`` otherwise. """ return cls._unpack(arg) if cls.test(arg) else None
@classmethod def _unpack(cls, arg): return arg.args __slots__ = ( '_args', '_annotations', '_hash', '_hexdigest', ) @abstractmethod def __init__(self, *args, **kwargs): self._init_cached() self._set_args(self._preprocess_args(args)) self._set_annotations(self._preprocess_annotations(kwargs)) self._hash = None self._hexdigest = None def _init_cached(self): pass def _set_args(self, args): self._args = args def _set_annotations(self, kwargs): self._annotations = kwargs def _preprocess_args(self, args): return tuple(map( self._preprocess_arg_callback, zip(args, util.count(1)))) def _preprocess_arg_callback(self, t): return self._preprocess_arg(*t) def _preprocess_arg(self, arg, i): return error.check_arg_is_not_none( arg, self.__class__.__name__, None, i) @staticmethod def _preprocess_arg_id(self, arg, i): return error.check_arg( arg, arg is not None and not Object.test(arg), 'invalid id', self.__class__.__name__, None, i, TypeError) def _preprocess_annotations(self, kwargs): return kwargs @property def args(self): """Object arguments.""" return self.get_args()
[docs] def get_args(self): """Gets object arguments. Returns: Object arguments. """ return self._args
@property def annotations(self): """Object annotations.""" return self.get_annotations()
[docs] def get_annotations(self): """Gets object annotations. Returns: Object annotations. """ return self._annotations
@property def hexdigest(self): """Object hexadecimal digest.""" return self.get_hexdigest()
[docs] def get_hexdigest(self): """Gets object hexadecimal digest. Returns: Object hexadecimal digest. """ if self._hexdigest is None: self._hexdigest = util.sha256( self.dump().encode('utf-8')).hexdigest() return self._hexdigest
def _as_id(self): """Generates an unique id for object. Returns: An unique id for object. """ return self._thy().settings.generated_id_prefix + self.hexdigest def __eq__(self, other): return type(self) == type(other) and self._args == other._args def __getitem__(self, i): return self.args[i] def __hash__(self): if self._hash is None: self._hash = hash((self.__class__, self._args)) return self._hash def __len__(self): return len(self.args) def __lt__(self, other): other = Object.check(other, '__lt__') if type(self) != type(other): return self.__class__.__name__ < other.__class__.__name__ else: return self.args < other.args def __matmul__(self, kwargs): return self.with_annotations(**kwargs) def __repr__(self): if self._thy().settings.override_object_repr: return str(self) else: return super().__repr__() def __str__(self): if self._thy().settings.serializer.default is not None: return self.serialize() else: return self.dump()
[docs] def dump(self): """Gets a raw string representation of object. Returns: A raw string representation of object. """ return self._dump()
def _dump(self, _f=(lambda x: x._dump() if Object.test(x) else str(x))): cls_name = self.__class__.__name__ if self.args: return f'({cls_name} {" ".join(map(_f, self.args))})' else: return cls_name # -- Comparison --------------------------------------------------------
[docs] def compare(self, other): """Compares object to `other`. Parameters: other: :class:`Object`. Returns: ``-1`` if object is less than `other`; ``0`` if object and `other` are equal; ``1`` if object is greater than `other`. See also: :meth:`Object.equal`. """ if self == other: return 0 elif self < other: return -1 else: return 1
[docs] def equal(self, other, deep=False): """Tests whether object is equal to `other`. Two objects are equal if they are instances of the same class and their arguments (:attr:`args`) are equal. If `deep` is ``True`` also compares the objects' :attr:`annotations` for equality. Parameters: other: :class:`Object`. deep: Whether to compare objects' annotations. Returns: ``True`` if successful; ``False`` otherwise. See also: :meth:`Object.deepequal`. """ if self != other: return False if not deep: return True if self.annotations != other.annotations: return False for x, y in zip(self, other): # compare args if isinstance(x, Object) and not x.equal(y, deep=deep): return False return True
[docs] def deepequal(self, other): """Tests whether object is deep-equal to `other`. Two objects are deep-equal if they are instances of the same class and their arguments (:attr:`args`) and annotations (:attr:`annotations`) are deep-equal. Parameters: other: :class:`Object`. Returns: ``True`` if successful; ``False`` otherwise. .. code-block:: python :caption: Equivalent to: obj.equal(other, deep=True) See also: :meth:`Object.equal`. """ return self.equal(other, deep=True)
# -- Copying -----------------------------------------------------------
[docs] def copy(self, *args, **kwargs): """Makes a shallow copy of object. If `args` are given, overwrites object arguments. If `kwargs` are given, overwrites object annotations. Parameters: args: Arguments. kwargs: Annotations. Returns: A shallow copy of object. See also: :meth:`Object.with_args`, :meth:`Object.with_annotations`. """ if not args and not kwargs: return util.copy(self) else: args = args or self.args kwargs = kwargs or self.annotations return self._dup(*args, **kwargs)
[docs] def with_args(self, *args): """Shallow-copies object overwriting its arguments. Parameters: args: Arguments. Returns: A shallow copy of object. .. code-block:: python :caption: Equivalent to: obj.copy(*args) See also: :meth:`Object.copy`. """ return self.copy(*args)
[docs] def with_annotations(self, **kwargs): """Shallow-copies object overwriting its annotations. Parameters: kwargs: Annotations. Returns: A shallow copy of object. .. code-block:: python :caption: Equivalent to: obj.copy(**kwargs) See also: :meth:`Object.copy`. """ return self.copy(**kwargs)
[docs] def deepcopy(self): """Makes a deep copy of object. Returns: A deep copy of object. """ return util.deepcopy(self)
# -- Conversion, parsing, serialization --------------------------------
[docs] @classmethod def convert_from(cls, value, format=None, **kwargs): """Converts `value` to object. Parameters: value: Value. format: Source format. kwargs: Options to converter. Returns: The resulting object. """ from .converter import Converter format = format or cls._thy().settings.converter.default format = error.check_arg_enum( format, Converter.converters, 'convert_from', 'format') return cls.check(Converter.convert_from( cls, value, format=format, **kwargs))
[docs] def convert_to(self, format=None, **kwargs): """Converts object to value. Parameters: format: Target format. kwargs: Options to converter. Returns: The resulting value. """ from .converter import Converter format = format or self._thy().settings.converter.default format = error.check_arg_enum( format, Converter.converters, 'convert_to', 'format') return Converter.convert_to(self, format=format, **kwargs)
[docs] @classmethod def parse( cls, from_=None, path=None, format=None, encoding=None, **kwargs): """Parses stream into object. Parameters: from_: Source stream. path: Source file path. format: Source format. encoding: Encoding. kwargs: Options to parser. Returns: The resulting object. """ from .parser import Parser format = format or cls._thy().settings.parser.default format = error.check_arg_enum( format, Parser.parsers, 'parse', 'format') return cls.check(Parser.parse( cls, from_=from_, path=path, format=format, encoding=encoding, **kwargs))
[docs] def serialize( self, to=None, path=None, format=None, encoding=None, **kwargs): """Serializes object to stream. Parameters: to: Target stream or file path. path: Target file path. format: Target format. encoding: Encoding. kwargs: Options to serializer. Returns: The resulting stream if `to` is ``None``. Otherwise, write the resulting stream to `to` and returns ``None``. """ from .serializer import Serializer format = format or self._thy().settings.serializer.default format = error.check_arg_enum( format, Serializer.serializers, 'serialize', 'format') return Serializer.serialize( self, to=to or path, format=format, encoding=encoding, **kwargs)
ObjectMeta._object_class = Object