dol.appendable¶
- Tools to add append-functionality to key-val stores. The main function is
appendable_store_cls = add_append_functionality_to_store_cls(store_cls, item2kv, …)
You give it the store_cls you want to sub class, and a item -> (key, val) function, and you get a store (subclass) that has a store.append(item) method. Also includes an extend method (that just called appends in a loop.
See add_append_functionality_to_store_cls docs for examples.
-
class
dol.appendable.
Extender
(store: collections.abc.MutableMapping, key, *, extend_store_value=<function read_add_write>, append_method=None)[source]¶ Extends a value in a store.
The value in the store (if it exists) must be an iterable. The value to extend must also be an iterable.
Unless a different
extend_store_value
function is given, the sum of the two iterables must be an iterable.The default
extend_store_value
is such that if the key is not in the store, the value is simply written in the store.The default
append_method
isNone
, which means that theappend
method is not defined. If you want to define it, you can pass a function that takes theExtender
instance as first argument, and the object to append as second argument. Theappend
method will then be defined as a partial of this function with theExtender
instance as first argument.>>> store = {'a': 'pple'} >>> # test normal extend >>> a_extender = Extender(store, 'a') >>> a_extender.extend('sauce') >>> store {'a': 'pplesauce'} >>> # test creation (when key is not in store) >>> b_extender = Extender(store, 'b') >>> b_extender.extend('anana') >>> store {'a': 'pplesauce', 'b': 'anana'} >>> # you can use the += operator too >>> b_extender += ' split' >>> store {'a': 'pplesauce', 'b': 'anana split'}
-
class
dol.appendable.
FirstAppendOnly
[source]¶ A finite Sequence that can have no more than one element.
>>> t = FirstAppendOnly() >>> assert len(t) == 0 >>> >>> t.append('something') >>> assert len(t) == 1 >>> assert t[0] == 'something' >>> >>> t.append('something else') >>> assert len(t) == 1 # still only one item >>> assert t[0] == 'something' # still the same item >>> >>> # Not that we'd ever these methods of FirstAppendOnly, but know that FirstAppendOnly is a collection.abc.Sequence, so... >>> t[:1] == t[:10] == t[::-1] == t[::-10] == t[0:2:10] == list(reversed(t)) == ['something'] True >>> >>> t.count('something') == 1 True >>> t.index('something') == 0 True
-
dol.appendable.
add_append_functionality_to_store_cls
(store_cls=None, *, item2kv, return_keys=False, __module__=None, __name__=None, __qualname__=None, __doc__=None, __annotations__=None, __defaults__=None, __kwdefaults__=None)¶ Makes a new class with append (and consequential extend) methods
- Parameters
store_cls – The store class to subclass
item2kv – The function that produces a (key, val) pair from an item
new_store_name – The name to give the new class (default will be ‘Appendable’ + store_cls.__name__)
Returns: A subclass of store_cls with two additional methods: append, and extend.
>>> item_to_kv = lambda item: (item['L'], item) # use value of 'L' as the key, and value is the item itself >>> MyStore = appendable(dict, item2kv=item_to_kv) >>> s = MyStore(); s.append({'L': 'let', 'I': 'it', 'G': 'go'}); list(s.items()) [('let', {'L': 'let', 'I': 'it', 'G': 'go'})]
Use mk_item2kv.from_item_to_key_params_and_val with tuple key params
>>> item_to_kv = appendable.mk_item2kv_for.item_to_key_params_and_val(lambda x: ((x['L'], x['I']), x['G']), '{}/{}') >>> MyStore = appendable(item2kv=item_to_kv)(dict) # showing the append(...)(store) form >>> s = MyStore(); s.append({'L': 'let', 'I': 'it', 'G': 'go'}); list(s.items()) [('let/it', 'go')]
Use mk_item2kv.from_item_to_key_params_and_val with dict key params
>>> item_to_kv = appendable.mk_item2kv_for.item_to_key_params_and_val( ... lambda x: ({'L': x['L'], 'G': x['G']}, x['I']), '{G}_{L}') >>> @appendable(item2kv=item_to_kv) # showing the @ form ... class MyStore(dict): ... pass >>> s = MyStore(); s.append({'L': 'let', 'I': 'it', 'G': 'go'}); list(s.items()) [('go_let', 'it')]
Use mk_item2kv.fields to get a tuple key from item fields, defining the sub-dict of the remaining fields to be the value. Also showing here how you can decorate the instance itself.
>>> item_to_kv = appendable.mk_item2kv_for.fields(['G', 'L'], key_as_tuple=True) >>> d = {} >>> s = appendable(d, item2kv=item_to_kv) >>> s.append({'L': 'let', 'I': 'it', 'G': 'go'}); list(s.items()) [(('go', 'let'), {'I': 'it'})]
You can make the “append” and “extend” methods to return the new generated keys by using the “return_keys” flag.
>>> d = {} >>> s = appendable(d, item2kv=item_to_kv, return_keys=True) >>> s.append({'L': 'let', 'I': 'it', 'G': 'go'}) ('go', 'let')
-
dol.appendable.
appendable
(store_cls=None, *, item2kv, return_keys=False, __module__=None, __name__=None, __qualname__=None, __doc__=None, __annotations__=None, __defaults__=None, __kwdefaults__=None)[source]¶ Makes a new class with append (and consequential extend) methods
- Parameters
store_cls – The store class to subclass
item2kv – The function that produces a (key, val) pair from an item
new_store_name – The name to give the new class (default will be ‘Appendable’ + store_cls.__name__)
Returns: A subclass of store_cls with two additional methods: append, and extend.
>>> item_to_kv = lambda item: (item['L'], item) # use value of 'L' as the key, and value is the item itself >>> MyStore = appendable(dict, item2kv=item_to_kv) >>> s = MyStore(); s.append({'L': 'let', 'I': 'it', 'G': 'go'}); list(s.items()) [('let', {'L': 'let', 'I': 'it', 'G': 'go'})]
Use mk_item2kv.from_item_to_key_params_and_val with tuple key params
>>> item_to_kv = appendable.mk_item2kv_for.item_to_key_params_and_val(lambda x: ((x['L'], x['I']), x['G']), '{}/{}') >>> MyStore = appendable(item2kv=item_to_kv)(dict) # showing the append(...)(store) form >>> s = MyStore(); s.append({'L': 'let', 'I': 'it', 'G': 'go'}); list(s.items()) [('let/it', 'go')]
Use mk_item2kv.from_item_to_key_params_and_val with dict key params
>>> item_to_kv = appendable.mk_item2kv_for.item_to_key_params_and_val( ... lambda x: ({'L': x['L'], 'G': x['G']}, x['I']), '{G}_{L}') >>> @appendable(item2kv=item_to_kv) # showing the @ form ... class MyStore(dict): ... pass >>> s = MyStore(); s.append({'L': 'let', 'I': 'it', 'G': 'go'}); list(s.items()) [('go_let', 'it')]
Use mk_item2kv.fields to get a tuple key from item fields, defining the sub-dict of the remaining fields to be the value. Also showing here how you can decorate the instance itself.
>>> item_to_kv = appendable.mk_item2kv_for.fields(['G', 'L'], key_as_tuple=True) >>> d = {} >>> s = appendable(d, item2kv=item_to_kv) >>> s.append({'L': 'let', 'I': 'it', 'G': 'go'}); list(s.items()) [(('go', 'let'), {'I': 'it'})]
You can make the “append” and “extend” methods to return the new generated keys by using the “return_keys” flag.
>>> d = {} >>> s = appendable(d, item2kv=item_to_kv, return_keys=True) >>> s.append({'L': 'let', 'I': 'it', 'G': 'go'}) ('go', 'let')
-
dol.appendable.
define_extend_as_seq_of_appends
(obj)[source]¶ Inject an extend method in obj that will used append method.
- Parameters
obj – Class (type) or instance of an object that has an “append” method.
Returns: The obj, but with that extend method.
>>> class A: ... def __init__(self): ... self.t = list() ... def append(self, item): ... self.t.append(item) ... >>> AA = define_extend_as_seq_of_appends(A) >>> a = AA() >>> a.extend([1,2,3]) >>> a.t [1, 2, 3] >>> a.extend([10, 20]) >>> a.t [1, 2, 3, 10, 20] >>> a = A() >>> a = define_extend_as_seq_of_appends(a) >>> a.extend([1,2,3]) >>> a.t [1, 2, 3] >>> a.extend([10, 20]) >>> a.t [1, 2, 3, 10, 20]
-
class
dol.appendable.
mk_item2kv_for
[source]¶ A bunch of functions to make item2kv functions
A few examples (see individual methods’ docs for more examples)
>>> # item_to_key >>> item2kv = mk_item2kv_for.item_to_key(item2key=lambda item: item['L'] ) >>> item2kv({'L': 'let', 'I': 'it', 'G': 'go'}) ('let', {'L': 'let', 'I': 'it', 'G': 'go'}) >>> >>> # utc_key >>> import time >>> item2key = mk_item2kv_for.utc_key() >>> k, v = item2key('some data') >>> assert abs(time.time() - k) < 0.01 # which asserts that k is indeed a (current) utc timestamp >>> assert v == 'some data' # just the item itself >>> >>> # item_to_key_params_and_val >>> item_to_kv = mk_item2kv_for.item_to_key_params_and_val(lambda x: ((x['L'], x['I']), x['G']), '{}/{}') >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) ('let/it', 'go') >>> >>> # fields >>> item_to_kv = mk_item2kv_for.fields(['L', 'I']) >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) ({'L': 'let', 'I': 'it'}, {'G': 'go'}) >>> item_to_kv = mk_item2kv_for.fields(('G', 'L'), keep_field_in_value=True) >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) # note the order of the key is not ('G', 'L')... ({'L': 'let', 'G': 'go'}, {'L': 'let', 'I': 'it', 'G': 'go'}) >>> item_to_kv = mk_item2kv_for.fields(('G', 'L'), key_as_tuple=True) # but ('G', 'L') order is respected here >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) (('go', 'let'), {'I': 'it'})
-
static
field
(field, keep_field_in_value=True, dflt_if_missing=<class 'dol.appendable.NotSpecified'>)[source]¶ item2kv that uses a specific key of a (mapping) item as the key
- Note: If keep_field_in_value=False, the field will be popped OUT of the item.
If that’s not the desired effect, one should feed copies of the items (e.g. map(dict.copy, items))
- Parameters
field – The field (value) to use as the returned key
keep_field_in_value – Whether to leave the field in the item. If False, will pop it out
dflt_if_missing – If specified (even None) will use the specified key as the key, if the field is missig
- Returns
A item2kv function
>>> item2kv = mk_item2kv_for.field('G') >>> item2kv({'L': 'let', 'I': 'it', 'G': 'go'}) ('go', {'L': 'let', 'I': 'it', 'G': 'go'}) >>> item2kv = mk_item2kv_for.field('G', keep_field_in_value=False) >>> item2kv({'L': 'let', 'I': 'it', 'G': 'go'}) ('go', {'L': 'let', 'I': 'it'}) >>> item2kv = mk_item2kv_for.field('G', dflt_if_missing=None) >>> item2kv({'L': 'let', 'I': 'it', 'DIE': 'go'}) (None, {'L': 'let', 'I': 'it', 'DIE': 'go'})
-
static
fields
(fields, keep_field_in_value=False, key_as_tuple=False)[source]¶ Make item2kv from specific fields of a Mapping (i.e. dict-like object) item.
Note: item2kv will not mutate item (even if keep_field_in_value=False).
- Parameters
fields – The sequence (list, tuple, etc.) of item fields that should be used to create the key.
keep_field_in_value – Set to True to return the item as is, as the value
key_as_tuple – Set to True if you want keys to be tuples (note that the fields order is important here!)
Returns: an item -> (item[fields], item[not in fields]) function
>>> item_to_kv = mk_item2kv_for.fields('L') >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) ({'L': 'let'}, {'I': 'it', 'G': 'go'}) >>> item_to_kv = mk_item2kv_for.fields(['L', 'I']) >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) ({'L': 'let', 'I': 'it'}, {'G': 'go'}) >>> item_to_kv = mk_item2kv_for.fields(('G', 'L'), keep_field_in_value=True) >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) # note the order of the key is not ('G', 'L')... ({'L': 'let', 'G': 'go'}, {'L': 'let', 'I': 'it', 'G': 'go'}) >>> item_to_kv = mk_item2kv_for.fields(('G', 'L'), key_as_tuple=True) # but ('G', 'L') order is respected here >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) (('go', 'let'), {'I': 'it'})
-
static
item_to_key
(item2key)[source]¶ Make item2kv from a item2key function (the value will be the item itself).
- Parameters
item2key – an item -> key function
Returns: an item -> (key, val) function
>>> item2key = lambda item: item['G'] # use value of 'L' as the key >>> item2key({'L': 'let', 'I': 'it', 'G': 'go'}) 'go' >>> item2kv = mk_item2kv_for.item_to_key(item2key) >>> item2kv({'L': 'let', 'I': 'it', 'G': 'go'}) ('go', {'L': 'let', 'I': 'it', 'G': 'go'})
-
static
item_to_key_params_and_val
(item_to_key_params_and_val, key_str_format)[source]¶ Make item2kv from a function that produces key_params and val, and a key_template that will produce a string key from the key_params
- Parameters
Returns: an item -> (key, val) function
>>> # Using tuple key params with unnamed string format fields >>> item_to_kv = mk_item2kv_for.item_to_key_params_and_val(lambda x: ((x['L'], x['I']), x['G']), '{}/{}') >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) ('let/it', 'go') >>> >>> # Using dict key params with named string format fields >>> item_to_kv = mk_item2kv_for.item_to_key_params_and_val( ... lambda x: ({'second': x['L'], 'first': x['G']}, x['I']), '{first}_{second}') >>> item_to_kv({'L': 'let', 'I': 'it', 'G': 'go'}) ('go_let', 'it')
-
static
utc_key
(offset_s=0, factor=1, *, time_postproc: Optional[Callable] = None)[source]¶ Make an item2kv function that uses the current time as the key, and the unchanged item as a value. The offset_s, which is added to the output key, can be used, for example, to align to another system’s clock, or to get a more accurate timestamp of an event.
- Use case for offset_s:
Align to another system’s clock
Get more accurate timestamping of an event. For example, in situations where the item is a chunk of live
streaming data and we want the key (timestamp) to represent the timestamp of the beginning of the chunk. Without an offset_s, the timestamp would be the timestamp after the last byte of the chunk was produced, plus the time it took to reach the present function. If we know the data production rate (e.g. sample rate) and the average lag to get to the present function, we can get a more accurate timestamp for the beginning of the chunk
- Parameters
offset_s – An offset (in seconds, possibly negative) to add to the current time.
Returns: an item -> (current_utc_s, item) function
>>> import time >>> item2key = mk_item2kv_for.utc_key() >>> k, v = item2key('some data') >>> assert abs(time.time() - k) < 0.01 # which asserts that k is indeed a (current) utc timestamp >>> assert v == 'some data' # just the item itself
-
static