Skip to content

deep.api.attributes

Check and create attributes.

BoundedAttributes

Bases: MutableMapping

An ordered dict with a fixed max capacity.

Oldest elements are dropped when the dict is full and a new element is added.

Source code in deep/api/attributes/__init__.py 
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
class BoundedAttributes(MutableMapping):
    """An ordered dict with a fixed max capacity.

    Oldest elements are dropped when the dict is full and a new element is
    added.
    """

    def __init__(
            self,
            max_length: Optional[int] = None,
            attributes: types.Attributes = None,
            immutable: bool = True,
            max_value_len: Optional[int] = None,
    ):
        """
        Create new attributes.

        :param max_length: max number of attributes
        :param attributes: existing attributes to copy
        :param immutable: are these attributes immutable
        :param max_value_len: max length of the attribute values
        """
        if max_length is not None:
            if not isinstance(max_length, int) or max_length < 0:
                raise ValueError(
                    "max_length must be valid int greater or equal to 0"
                )
        self.max_length = max_length
        self.dropped = 0
        self.max_value_len = max_value_len
        self._dict = OrderedDict()  # type: OrderedDict
        self._lock = threading.Lock()  # type: threading.Lock
        if attributes:
            for key, value in attributes.items():
                self[key] = value
        self._immutable = immutable

    def __repr__(self):
        """Represent this as a string."""
        return (
            f"{type(self).__name__}({dict(self._dict)}, max_length={self.max_length})"
        )

    def __getitem__(self, key):
        """Get attribute value."""
        return self._dict[key]

    def __setitem__(self, key, value):
        """Set attribute value."""
        if getattr(self, "_immutable", False):
            raise TypeError
        with self._lock:
            if self.max_length is not None and self.max_length == 0:
                self.dropped += 1
                return

            value = _clean_attribute(key, value, self.max_value_len)
            if value is not None:
                if key in self._dict:
                    del self._dict[key]
                elif (
                        self.max_length is not None and len(self._dict) == self.max_length
                ):
                    self._dict.popitem(last=False)
                    self.dropped += 1

                self._dict[key] = value

    def __delitem__(self, key):
        """Delete item from attributes."""
        if getattr(self, "_immutable", False):
            raise TypeError
        with self._lock:
            del self._dict[key]

    def __iter__(self):
        """Create iterator."""
        with self._lock:
            return iter(self._dict.copy())

    def __len__(self):
        """Get number of attributes."""
        return len(self._dict)

    def copy(self):
        """Create a copy of these attributes."""
        return self._dict.copy()

    def merge_in(self, attributes):
        """Merge in another attributes object."""
        for k, v in attributes.items():
            self[k] = v

__delitem__(key)

Delete item from attributes.

Source code in deep/api/attributes/__init__.py 
191
192
193
194
195
196
def __delitem__(self, key):
    """Delete item from attributes."""
    if getattr(self, "_immutable", False):
        raise TypeError
    with self._lock:
        del self._dict[key]

__getitem__(key)

Get attribute value.

Source code in deep/api/attributes/__init__.py 
166
167
168
def __getitem__(self, key):
    """Get attribute value."""
    return self._dict[key]

__init__(max_length=None, attributes=None, immutable=True, max_value_len=None)

Create new attributes.

:param max_length: max number of attributes :param attributes: existing attributes to copy :param immutable: are these attributes immutable :param max_value_len: max length of the attribute values

Source code in deep/api/attributes/__init__.py 
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
def __init__(
        self,
        max_length: Optional[int] = None,
        attributes: types.Attributes = None,
        immutable: bool = True,
        max_value_len: Optional[int] = None,
):
    """
    Create new attributes.

    :param max_length: max number of attributes
    :param attributes: existing attributes to copy
    :param immutable: are these attributes immutable
    :param max_value_len: max length of the attribute values
    """
    if max_length is not None:
        if not isinstance(max_length, int) or max_length < 0:
            raise ValueError(
                "max_length must be valid int greater or equal to 0"
            )
    self.max_length = max_length
    self.dropped = 0
    self.max_value_len = max_value_len
    self._dict = OrderedDict()  # type: OrderedDict
    self._lock = threading.Lock()  # type: threading.Lock
    if attributes:
        for key, value in attributes.items():
            self[key] = value
    self._immutable = immutable

__iter__()

Create iterator.

Source code in deep/api/attributes/__init__.py 
198
199
200
201
def __iter__(self):
    """Create iterator."""
    with self._lock:
        return iter(self._dict.copy())

__len__()

Get number of attributes.

Source code in deep/api/attributes/__init__.py 
203
204
205
def __len__(self):
    """Get number of attributes."""
    return len(self._dict)

__repr__()

Represent this as a string.

Source code in deep/api/attributes/__init__.py 
160
161
162
163
164
def __repr__(self):
    """Represent this as a string."""
    return (
        f"{type(self).__name__}({dict(self._dict)}, max_length={self.max_length})"
    )

__setitem__(key, value)

Set attribute value.

Source code in deep/api/attributes/__init__.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def __setitem__(self, key, value):
    """Set attribute value."""
    if getattr(self, "_immutable", False):
        raise TypeError
    with self._lock:
        if self.max_length is not None and self.max_length == 0:
            self.dropped += 1
            return

        value = _clean_attribute(key, value, self.max_value_len)
        if value is not None:
            if key in self._dict:
                del self._dict[key]
            elif (
                    self.max_length is not None and len(self._dict) == self.max_length
            ):
                self._dict.popitem(last=False)
                self.dropped += 1

            self._dict[key] = value

copy()

Create a copy of these attributes.

Source code in deep/api/attributes/__init__.py 
207
208
209
def copy(self):
    """Create a copy of these attributes."""
    return self._dict.copy()

merge_in(attributes)

Merge in another attributes object.

Source code in deep/api/attributes/__init__.py 
211
212
213
214
def merge_in(self, attributes):
    """Merge in another attributes object."""
    for k, v in attributes.items():
        self[k] = v