forked from basho/riak-python-client
-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathset.py
More file actions
131 lines (100 loc) · 3.55 KB
/
Copy pathset.py
File metadata and controls
131 lines (100 loc) · 3.55 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
# Copyright 2010-present Basho Technologies, Inc.
#
# Licensed 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.
import collections
from riak.datatypes import TYPES
from .datatype import Datatype
__all__ = ["Set"]
class Set(collections.Set, Datatype):
"""A convergent datatype representing a Set with observed-remove
semantics. Currently strings are the only supported value type.
Example::
myset.add("barista")
myset.add("roaster")
myset.add("brewer")
Likewise they can simply be removed::
myset.discard("barista")
This datatype also implements the `Set ABC
<https://docs.python.org/2/library/collections.html>`_, meaning it
supports ``len()``, ``in``, and iteration.
"""
type_name = "set"
_type_error_msg = "Sets can only be iterables of strings"
def _post_init(self):
self._adds = set()
self._removes = set()
def _default_value(self):
return frozenset()
@Datatype.modified.getter
def modified(self):
"""
Whether this set has staged adds or removes.
"""
return len(self._removes | self._adds) > 0
def to_op(self):
"""
Extracts the modification operation from the set.
:rtype: dict, None
"""
if not self._adds and not self._removes:
return None
changes = {}
if self._adds:
changes["adds"] = list(self._adds)
if self._removes:
changes["removes"] = list(self._removes)
return changes
# collections.Set API, operates only on the immutable version
def __contains__(self, element):
return element in self.value
def __iter__(self):
return iter(self.value)
def __len__(self):
return len(self.value)
# Sort of like collections.MutableSet API, without the additional
# methods.
def add(self, element):
"""
Adds an element to the set.
.. note: You may add elements that already exist in the set.
This may be used as an "assertion" that the element is a
member.
:param element: the element to add
:type element: str
"""
_check_element(element)
self._adds.add(element)
def discard(self, element):
"""
Removes an element from the set.
.. note: You may remove elements from the set that are not
present, but a context from the server is required.
:param element: the element to remove
:type element: str
"""
_check_element(element)
self._require_context()
self._removes.add(element)
def _coerce_value(self, new_value):
return frozenset(new_value)
def _check_type(self, new_value):
if not isinstance(new_value, collections.Iterable):
return False
for element in new_value:
if not isinstance(element, str):
return False
return True
def _check_element(element):
if not isinstance(element, str):
raise TypeError("Set elements can only be strings")
TYPES["set"] = Set