Recipes

Data peeks

Print db and collection names of local mongo instances

import mongodol as m

for db_name in m.MongoClientReader():
    print(f"{db_name}:\n\t{', '.join((m.MongoDbReader(db_name)))}\n")

# admin:
# 	system.version
# 
# config:
# 	system.sessions
# 
# local:
# 	startup_log
# 
# mongodol:
# 	mongodol_test, test, bdfl, number, feature_cube, tmp
# 
# ...

Print db and collection names with number of docs of each

import mongodol as m

client = m.MongoClientReader()
for db_name in client:
    db = client[db_name]
    collections_with_size = [f"{collection_name} ({len(db[collection_name])})" for collection_name in db]
    print(f"{db_name}:\n\t{', '.join(collections_with_size)}\n")

# admin:
# 	system.version (1)
# 
# config:
# 	system.sessions (45)
# 
# local:
# 	startup_log (40)
# 
# mongodol:
# 	mongodol_test (1), test (0), bdfl (33), number (5), feature_cube (7), tmp (0)
# 
# ...

Misc

from operator import itemgetter
from functools import partial

from mongodol.stores import MongoTupleKeyStore
from dol import wrap_kvs

def itemsetter(item, key, container_factory=dict):
    """A (single item) dual of itemgetter. To call repeatedly on the same container, use lambda: mutable_container_reference"""
    container = container_factory()
    container[key] = item
    return container

# Note: Could also add input type validation in data_of_obj
wrapper = wrap_kvs(
    data_of_obj=partial(itemsetter, key='wf'),
    obj_of_data=itemgetter('wf')
)
s = wrapper(MongoTupleKeyStore('mongodol', 'test', key_fields=('session', 'block'), data_fields=['wf']))

for k in s:
    del s[k]

assert list(s) == []

s[123, 45] = [1, 2, 3]
s[42, 24] = [4, 5, 6, 7]

assert sorted(s) == [(42, 24), (123, 45)]
assert s[123, 45] == [1, 2, 3]
assert s[42, 24] == [4, 5, 6, 7]