forked from apache/ignite
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbinary.py
More file actions
136 lines (117 loc) · 5.01 KB
/
Copy pathbinary.py
File metadata and controls
136 lines (117 loc) · 5.01 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
# 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
from typing import Any
import attr
from .datatypes import *
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 GenericObjectPropsMixin:
"""
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 type_name(self) -> str:
""" Binary object type name. """
return self._type_name
@property
def type_id(self) -> int:
""" Binary object type ID. """
return entity_id(self._type_name)
@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 cls is not GenericObjectMeta:
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 GenericObjectMeta(type, GenericObjectPropsMixin):
"""
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. """
return super().__new__(
mcs, name, (GenericObjectPropsMixin, )+base_classes, namespace
)
@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)