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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
|
from pyrsistent._checked_types import (InvariantException, CheckedType, _restore_pickle, store_invariants)
from pyrsistent._field_common import (
set_fields, check_type, is_field_ignore_extra_complaint, PFIELD_NO_INITIAL, serialize, check_global_invariants
)
from pyrsistent._transformations import transform
def _is_pclass(bases):
return len(bases) == 1 and bases[0] == CheckedType
class PClassMeta(type):
def __new__(mcs, name, bases, dct):
set_fields(dct, bases, name='_pclass_fields')
store_invariants(dct, bases, '_pclass_invariants', '__invariant__')
dct['__slots__'] = ('_pclass_frozen',) + tuple(key for key in dct['_pclass_fields'])
# There must only be one __weakref__ entry in the inheritance hierarchy,
# lets put it on the top level class.
if _is_pclass(bases):
dct['__slots__'] += ('__weakref__',)
return super(PClassMeta, mcs).__new__(mcs, name, bases, dct)
_MISSING_VALUE = object()
def _check_and_set_attr(cls, field, name, value, result, invariant_errors):
check_type(cls, field, name, value)
is_ok, error_code = field.invariant(value)
if not is_ok:
invariant_errors.append(error_code)
else:
setattr(result, name, value)
class PClass(CheckedType, metaclass=PClassMeta):
"""
A PClass is a python class with a fixed set of specified fields. PClasses are declared as python classes inheriting
from PClass. It is defined the same way that PRecords are and behaves like a PRecord in all aspects except that it
is not a PMap and hence not a collection but rather a plain Python object.
More documentation and examples of PClass usage is available at https://github.com/tobgu/pyrsistent
"""
def __new__(cls, **kwargs): # Support *args?
result = super(PClass, cls).__new__(cls)
factory_fields = kwargs.pop('_factory_fields', None)
ignore_extra = kwargs.pop('ignore_extra', None)
missing_fields = []
invariant_errors = []
for name, field in cls._pclass_fields.items():
if name in kwargs:
if factory_fields is None or name in factory_fields:
if is_field_ignore_extra_complaint(PClass, field, ignore_extra):
value = field.factory(kwargs[name], ignore_extra=ignore_extra)
else:
value = field.factory(kwargs[name])
else:
value = kwargs[name]
_check_and_set_attr(cls, field, name, value, result, invariant_errors)
del kwargs[name]
elif field.initial is not PFIELD_NO_INITIAL:
initial = field.initial() if callable(field.initial) else field.initial
_check_and_set_attr(
cls, field, name, initial, result, invariant_errors)
elif field.mandatory:
missing_fields.append('{0}.{1}'.format(cls.__name__, name))
if invariant_errors or missing_fields:
raise InvariantException(tuple(invariant_errors), tuple(missing_fields), 'Field invariant failed')
if kwargs:
raise AttributeError("'{0}' are not among the specified fields for {1}".format(
', '.join(kwargs), cls.__name__))
check_global_invariants(result, cls._pclass_invariants)
result._pclass_frozen = True
return result
def set(self, *args, **kwargs):
"""
Set a field in the instance. Returns a new instance with the updated value. The original instance remains
unmodified. Accepts key-value pairs or single string representing the field name and a value.
>>> from pyrsistent import PClass, field
>>> class AClass(PClass):
... x = field()
...
>>> a = AClass(x=1)
>>> a2 = a.set(x=2)
>>> a3 = a.set('x', 3)
>>> a
AClass(x=1)
>>> a2
AClass(x=2)
>>> a3
AClass(x=3)
"""
if args:
kwargs[args[0]] = args[1]
factory_fields = set(kwargs)
for key in self._pclass_fields:
if key not in kwargs:
value = getattr(self, key, _MISSING_VALUE)
if value is not _MISSING_VALUE:
kwargs[key] = value
return self.__class__(_factory_fields=factory_fields, **kwargs)
@classmethod
def create(cls, kwargs, _factory_fields=None, ignore_extra=False):
"""
Factory method. Will create a new PClass of the current type and assign the values
specified in kwargs.
:param ignore_extra: A boolean which when set to True will ignore any keys which appear in kwargs that are not
in the set of fields on the PClass.
"""
if isinstance(kwargs, cls):
return kwargs
if ignore_extra:
kwargs = {k: kwargs[k] for k in cls._pclass_fields if k in kwargs}
return cls(_factory_fields=_factory_fields, ignore_extra=ignore_extra, **kwargs)
def serialize(self, format=None):
"""
Serialize the current PClass using custom serializer functions for fields where
such have been supplied.
"""
result = {}
for name in self._pclass_fields:
value = getattr(self, name, _MISSING_VALUE)
if value is not _MISSING_VALUE:
result[name] = serialize(self._pclass_fields[name].serializer, format, value)
return result
def transform(self, *transformations):
"""
Apply transformations to the currency PClass. For more details on transformations see
the documentation for PMap. Transformations on PClasses do not support key matching
since the PClass is not a collection. Apart from that the transformations available
for other persistent types work as expected.
"""
return transform(self, transformations)
def __eq__(self, other):
if isinstance(other, self.__class__):
for name in self._pclass_fields:
if getattr(self, name, _MISSING_VALUE) != getattr(other, name, _MISSING_VALUE):
return False
return True
return NotImplemented
def __ne__(self, other):
return not self == other
def __hash__(self):
# May want to optimize this by caching the hash somehow
return hash(tuple((key, getattr(self, key, _MISSING_VALUE)) for key in self._pclass_fields))
def __setattr__(self, key, value):
if getattr(self, '_pclass_frozen', False):
raise AttributeError("Can't set attribute, key={0}, value={1}".format(key, value))
super(PClass, self).__setattr__(key, value)
def __delattr__(self, key):
raise AttributeError("Can't delete attribute, key={0}, use remove()".format(key))
def _to_dict(self):
result = {}
for key in self._pclass_fields:
value = getattr(self, key, _MISSING_VALUE)
if value is not _MISSING_VALUE:
result[key] = value
return result
def __repr__(self):
return "{0}({1})".format(self.__class__.__name__,
', '.join('{0}={1}'.format(k, repr(v)) for k, v in self._to_dict().items()))
def __reduce__(self):
# Pickling support
data = dict((key, getattr(self, key)) for key in self._pclass_fields if hasattr(self, key))
return _restore_pickle, (self.__class__, data,)
def evolver(self):
"""
Returns an evolver for this object.
"""
return _PClassEvolver(self, self._to_dict())
def remove(self, name):
"""
Remove attribute given by name from the current instance. Raises AttributeError if the
attribute doesn't exist.
"""
evolver = self.evolver()
del evolver[name]
return evolver.persistent()
class _PClassEvolver(object):
__slots__ = ('_pclass_evolver_original', '_pclass_evolver_data', '_pclass_evolver_data_is_dirty', '_factory_fields')
def __init__(self, original, initial_dict):
self._pclass_evolver_original = original
self._pclass_evolver_data = initial_dict
self._pclass_evolver_data_is_dirty = False
self._factory_fields = set()
def __getitem__(self, item):
return self._pclass_evolver_data[item]
def set(self, key, value):
if self._pclass_evolver_data.get(key, _MISSING_VALUE) is not value:
self._pclass_evolver_data[key] = value
self._factory_fields.add(key)
self._pclass_evolver_data_is_dirty = True
return self
def __setitem__(self, key, value):
self.set(key, value)
def remove(self, item):
if item in self._pclass_evolver_data:
del self._pclass_evolver_data[item]
self._factory_fields.discard(item)
self._pclass_evolver_data_is_dirty = True
return self
raise AttributeError(item)
def __delitem__(self, item):
self.remove(item)
def persistent(self):
if self._pclass_evolver_data_is_dirty:
return self._pclass_evolver_original.__class__(_factory_fields=self._factory_fields,
**self._pclass_evolver_data)
return self._pclass_evolver_original
def __setattr__(self, key, value):
if key not in self.__slots__:
self.set(key, value)
else:
super(_PClassEvolver, self).__setattr__(key, value)
def __getattr__(self, item):
return self[item]
|
