pyignite/binary.py

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
:class:`~pyignite.binary.GenericObjectMeta` is a metaclass used to create
classes, which objects serve as a native Python values for Ignite Complex
object data type. You can use this metaclass with your existing classes
to save and restore their selected attributes and properties to/from
Ignite caches. It is also used internally by `pyignite` to create simple
data classes “on the fly” when retrieving arbitrary Complex objects.

You can get the examples of using Complex objects in the
:ref:`complex_object_usage` section of `pyignite` documentation.
"""

from collections import OrderedDict
import ctypes
from io import SEEK_CUR
from typing import Any

import attr

from .constants import PROTOCOL_BYTE_ORDER
from .datatypes import (
    Null, ByteObject, ShortObject, IntObject, LongObject, FloatObject, DoubleObject, CharObject, BoolObject, UUIDObject,
    DateObject, TimestampObject, TimeObject, EnumObject, BinaryEnumObject, ByteArrayObject, ShortArrayObject,
    IntArrayObject, LongArrayObject, FloatArrayObject, DoubleArrayObject, CharArrayObject, BoolArrayObject,
    UUIDArrayObject, DateArrayObject, TimestampArrayObject, TimeArrayObject, EnumArrayObject, String, StringArrayObject,
    DecimalObject, DecimalArrayObject, ObjectArrayObject, CollectionObject, MapObject, BinaryObject, WrappedDataObject
)
from .datatypes.base import IgniteDataTypeProps
from .exceptions import ParseError
from .utils import entity_id, schema_id


ALLOWED_FIELD_TYPES = [
    Null, ByteObject, ShortObject, IntObject, LongObject, FloatObject,
    DoubleObject, CharObject, BoolObject, UUIDObject, DateObject,
    TimestampObject, TimeObject, EnumObject, BinaryEnumObject,
    ByteArrayObject, ShortArrayObject, IntArrayObject, LongArrayObject,
    FloatArrayObject, DoubleArrayObject, CharArrayObject, BoolArrayObject,
    UUIDArrayObject, DateArrayObject, TimestampArrayObject,
    TimeArrayObject, EnumArrayObject, String, StringArrayObject,
    DecimalObject, DecimalArrayObject, ObjectArrayObject, CollectionObject,
    MapObject, BinaryObject, WrappedDataObject,
]


class GenericObjectProps(IgniteDataTypeProps):
    """
    This class is mixed both to metaclass and to resulting class to make class
    properties universally available. You should not subclass it directly.
    """
    @property
    def schema(self) -> OrderedDict:
        """ Binary object schema. """
        return self._schema.copy()

    @property
    def schema_id(self) -> int:
        """ Binary object schema ID. """
        return schema_id(self._schema)

    def __new__(cls, *args, **kwargs) -> Any:
        # allow all items in Binary Object schema to be populated as optional
        # arguments to `__init__()` with sensible defaults.
        if not attr.has(cls):
            attributes = {
                k: attr.ib(type=getattr(v, 'pythonic', type(None)), default=getattr(v, 'default', None))
                for k, v in cls.schema.items()
            }

            attributes.update({'version': attr.ib(type=int, default=1)})
            cls = attr.s(cls, these=attributes)
        # skip parameters
        return super().__new__(cls)


class GenericObjectPropsMeta(type, GenericObjectProps):
    pass


class GenericObjectMeta(GenericObjectPropsMeta):
    """
    Complex (or Binary) Object metaclass. It is aimed to help user create
    classes, which objects could serve as a pythonic representation of the
    :class:`~pyignite.datatypes.complex.BinaryObject` Ignite data type.
    """
    _schema = None
    _type_name = None
    version = None

    def __new__(
        mcs: Any, name: str, base_classes: tuple, namespace: dict, **kwargs
    ) -> Any:
        """ Sort out class creation arguments. """

        result = super().__new__(
            mcs, name, (GenericObjectProps, ) + base_classes, namespace
        )

        def _from_python(self, stream, save_to_buf=False):
            """
            Method for building binary representation of the Generic object
            and calculating a hashcode from it.

            :param self: Generic object instance,
            :param stream: BinaryStream
            :param save_to_buf: Optional. If True, save serialized data to buffer.
            """
            initial_pos = stream.tell()
            header, header_class = write_header(self, stream)

            offsets = [ctypes.sizeof(header_class)]
            schema_items = list(self.schema.items())
            for field_name, field_type in schema_items:
                val = getattr(self, field_name, getattr(field_type, 'default', None))
                field_start_pos = stream.tell()
                field_type.from_python(stream, val)
                offsets.append(max(offsets) + stream.tell() - field_start_pos)

            write_footer(self, stream, header, header_class, schema_items, offsets, initial_pos, save_to_buf)

        async def _from_python_async(self, stream, save_to_buf=False):
            """
            Async version of _from_python
            """
            initial_pos = stream.tell()
            header, header_class = write_header(self, stream)

            offsets = [ctypes.sizeof(header_class)]
            schema_items = list(self.schema.items())
            for field_name, field_type in schema_items:
                val = getattr(self, field_name, getattr(field_type, 'default', None))
                field_start_pos = stream.tell()
                await field_type.from_python_async(stream, val)
                offsets.append(max(offsets) + stream.tell() - field_start_pos)

            write_footer(self, stream, header, header_class, schema_items, offsets, initial_pos, save_to_buf)

        def write_header(obj, stream):
            header_class = BinaryObject.get_header_class()
            header = header_class()
            header.type_code = int.from_bytes(
                BinaryObject.type_code,
                byteorder=PROTOCOL_BYTE_ORDER
            )
            header.flags = BinaryObject.USER_TYPE | BinaryObject.HAS_SCHEMA
            if stream.compact_footer:
                header.flags |= BinaryObject.COMPACT_FOOTER
            header.version = obj.version
            header.type_id = obj.type_id
            header.schema_id = obj.schema_id

            stream.seek(ctypes.sizeof(header_class), SEEK_CUR)

            return header, header_class

        def write_footer(obj, stream, header, header_class, schema_items, offsets, initial_pos, save_to_buf):
            offsets = offsets[:-1]
            header_len = ctypes.sizeof(header_class)

            # create footer
            if max(offsets, default=0) < 255:
                header.flags |= BinaryObject.OFFSET_ONE_BYTE
            elif max(offsets) < 65535:
                header.flags |= BinaryObject.OFFSET_TWO_BYTES

            schema_class = BinaryObject.schema_type(header.flags) * len(offsets)
            schema = schema_class()

            if stream.compact_footer:
                for i, offset in enumerate(offsets):
                    schema[i] = offset
            else:
                for i, offset in enumerate(offsets):
                    schema[i].field_id = entity_id(schema_items[i][0])
                    schema[i].offset = offset

            # calculate size and hash code
            fields_data_len = stream.tell() - initial_pos - header_len
            header.schema_offset = fields_data_len + header_len
            header.length = header.schema_offset + ctypes.sizeof(schema_class)
            header.hash_code = stream.hashcode(initial_pos + header_len, fields_data_len)

            stream.seek(initial_pos)
            stream.write(header)
            stream.seek(initial_pos + header.schema_offset)
            stream.write(schema)

            if save_to_buf:
                obj._buffer = stream.slice(initial_pos, stream.tell() - initial_pos)
            obj._hashcode = header.hash_code

        def _setattr(self, attr_name: str, attr_value: Any):
            # reset binary representation, if any field is changed
            if attr_name in self._schema.keys():
                self._buffer = None
                self._hashcode = None

            # `super()` is really need these parameters
            super(result, self).__setattr__(attr_name, attr_value)

        setattr(result, _from_python.__name__, _from_python)
        setattr(result, _from_python_async.__name__, _from_python_async)
        setattr(result, '__setattr__', _setattr)
        setattr(result, '_buffer', None)
        setattr(result, '_hashcode', None)
        return result

    @staticmethod
    def _validate_schema(schema: dict):
        for field_type in schema.values():
            if field_type not in ALLOWED_FIELD_TYPES:
                raise ParseError(
                    'Wrong binary field type: {}'.format(field_type)
                )

    def __init__(
        cls, name: str, base_classes: tuple, namespace: dict,
        type_name: str = None, schema: OrderedDict = None, **kwargs
    ):
        """
        Initializes binary object class.

        :param type_name: (optional) binary object name. Defaults to class
         name,
        :param schema: (optional) a dict of field names: field types,
        :raise: ParseError if one or more binary field types
         did not recognized.
        """
        cls._type_name = type_name or cls.__name__
        cls._type_id = entity_id(cls._type_name)
        schema = schema or OrderedDict()
        cls._validate_schema(schema)
        cls._schema = schema
        super().__init__(name, base_classes, namespace)