davidlatwe / montydb / 211

pending completion

Build # 211

Build Type

Pull #18

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/davidlatwe/montydb/commit/<a class=hub.com/davidlatwe/montydb/commit/<a class="double-link" href="https://git"><a class=hub.com/davidlatwe/montydb/commit/<a class="double-link" href="https://git"><a class=hub.com/davidlatwe/montydb/commit/03ccda231af2d8da820b7685c88aca766852f8ec">03ccda231<a href="https://github.com/davidlatwe/montydb/commit/03ccda231af2d8da820b7685c88aca766852f8ec">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/davidlatwe/montydb/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/davidlatwe/montydb/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/davidlatwe/montydb/commit/03ccda231af2d8da820b7685c88aca766852f8ec&quot;&gt;03ccda231&lt;/a&gt;&lt;a href=&quot;https://github.com/davidlatwe/montydb/commit/03ccda231af2d8da820b7685c88aca766852f8ec&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/davidlatwe/montydb/commit/03ccda231af2d8da820b7685c88aca76&lt;/a&gt;6852f8ec">&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/davidlatwe/montydb/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/davidlatwe/montydb/commit/<a class="double-link" href="https://github.com/davidlatwe/montydb/commit/e7237018ac2d7f4fe3fba20453821c2dbb38d2b2">e7237018a</a><a href="https://github.com/davidlatwe/montydb/commit/03ccda231af2d8da820b7685c88aca766852f8ec">&amp;quot;&amp;gt;e7237018a&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;e7237018a&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/davidlatwe/montydb/commit/03ccda231af2d8da820b7685c88aca766852f&lt;/a&gt;8ec&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/davidlatwe/montydb/commit/<a class="double-link" href="https://github.com/davidlatwe/montydb/commit/5919826d6">5919826d6">5919826d6</a>

Pull Request Pull Request #18: Implement LMDB storage

Run Details

1289 of 1574 branches covered (81.89%)

Branch coverage included in aggregate %.

172 of 172 new or added lines in 3 files covered. (100.0%)

3436 of 3898 relevant lines covered (88.15%)

13.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/montydb/storage/lightning.py


import os
import lmdb
import shutil
from itertools import islice

from ..types import unicode_
from . import (
    AbstractStorage,
    AbstractDatabase,
    AbstractCollection,
    AbstractCursor,

    StorageDuplicateKeyError,
)

# Functional > Performence > Natural

LMDB_DB_EXT = ".mdb"


def _to_bytes(s):
    return s.encode()


class LMDBKVEngine(object):
    """Per collection"""

    dbname = _to_bytes("documents")

    def __init__(self, options=None):
        """
        """
        opt = options or dict()
        opt.update({
            "subdir": False,
            "max_dbs": 1,
        })
        self.opt = opt
        self._path = None

    def set_path(self, path):
        self._path = path

    def open(self):
        environment = lmdb.open(self._path, **self.opt)
        return environment

    def iter_docs(self, environment):
        if not os.path.isfile(self._path):
            return

        with environment as env:
            db = env.open_db(self.dbname)
            with env.begin(db, write=False) as txn:
                cursor = txn.cursor()
                for encoded_doc in cursor.iternext(keys=False, values=True):
                    yield encoded_doc

    def write(self, environment, pairs, overwrite=False):
        if not os.path.isfile(self._path):
            return

        dup = False
        with environment as env:
            db = env.open_db(self.dbname)
            with env.begin(db, write=True) as txn:
                for doc_id, encoded_doc in pairs:
                    id = _to_bytes(str(doc_id))
                    if not txn.put(id, encoded_doc, overwrite=overwrite):
                        dup = True
                        break
        if dup:
            raise StorageDuplicateKeyError()

    def delete(self, environment, doc_ids):
        if not os.path.isfile(self._path):
            return

        with environment as env:
            db = env.open_db(self.dbname, reverse_key=False)
            with env.begin(db, write=True) as txn:
                cursor = txn.cursor()
                for doc_id in doc_ids:
                    id = _to_bytes(str(doc_id))
                    if cursor.set_key(id):
                        cursor.delete()


class LMDBStorage(AbstractStorage):
    """
    """

    def __init__(self, repository, storage_config):
        super(LMDBStorage, self).__init__(repository, storage_config)
        self._conn = LMDBKVEngine(self._config)

    def _db_path(self, db_name):
        """
        Get Monty database dir path.
        """
        return os.path.join(self._repository, db_name)

    @classmethod
    def nice_name(cls):
        return "lightning"

    @classmethod
    def config(cls, map_size=10485760, **kwargs):
        """
        """
        return {
            "map_size": int(map_size),
        }

    def database_create(self, db_name):
        if not os.path.isdir(self._db_path(db_name)):
            os.makedirs(self._db_path(db_name))

    def database_drop(self, db_name):
        db_path = self._db_path(db_name)
        if os.path.isdir(db_path):
            shutil.rmtree(db_path)

    def database_list(self):
        return [
            name for name in os.listdir(unicode_(self._repository))
            if os.path.isdir(self._db_path(name))
        ]


class LMDBDatabase(AbstractDatabase):
    """
    """

    def __init__(self, storage, subject):
        super(LMDBDatabase, self).__init__(storage, subject)
        self._db_path = storage._db_path(self._name)
        self._conn = storage._conn

    def _col_path(self, col_name):
        """
        Get database file path, which is Monty collection.
        """
        return os.path.join(self._db_path, col_name) + LMDB_DB_EXT

    def database_exists(self):
        return os.path.isdir(self._db_path)

    def collection_exists(self, col_name):
        return os.path.isfile(self._col_path(col_name))

    def collection_create(self, col_name):
        if not self.database_exists():
            self._storage.database_create(self._name)
        self._conn.set_path(self._col_path(col_name))
        environment = self._conn.open()

        return environment

    def collection_drop(self, col_name):
        if self.collection_exists(col_name):
            os.remove(self._col_path(col_name))

    def collection_list(self):
        if not self.database_exists():
            return []
        return [os.path.splitext(name)[0]
                for name in os.listdir(unicode_(self._db_path))
                if name.endswith(LMDB_DB_EXT)]


LMDBStorage.contractor_cls = LMDBDatabase


class LMDBCollection(AbstractCollection):
    """
    """

    def __init__(self, database, subject):
        super(LMDBCollection, self).__init__(database, subject)
        self._conn = database._conn
        self._conn.set_path(database._col_path(self._name))

    def _ensure_table(func):
        def make_table(self, *args, **kwargs):
            environment = None
            if not self._database.collection_exists(self._name):
                environment = self._database.collection_create(self._name)
            return func(self, env=environment, *args, **kwargs)
        return make_table

    @_ensure_table
    def write_one(self, doc, check_keys=True, env=None):
        env = env or self._conn.open()

        id = doc["_id"]
        encoded = self._encode_doc(doc, check_keys)
        self._conn.write(env, [(id, encoded)])

        return id

    @_ensure_table
    def write_many(self, docs, check_keys=True, ordered=True, env=None):
        env = env or self._conn.open()
        ids = list()

        def produce_encoded_docs():
            for doc in docs:
                id = doc["_id"]
                yield id, self._encode_doc(doc, check_keys)
                ids.append(id)

        self._conn.write(env, produce_encoded_docs())

        return ids

    def update_one(self, doc):
        env = self._conn.open()

        id = doc["_id"]
        encoded = self._encode_doc(doc)
        self._conn.write(env, [(id, encoded)], overwrite=True)

    def update_many(self, docs):
        env = self._conn.open()

        def produce_encoded_docs():
            for doc in docs:
                yield doc["_id"], self._encode_doc(doc)

        self._conn.write(env, produce_encoded_docs(), overwrite=True)

    def delete_one(self, id):
        env = self._conn.open()
        self._conn.delete(env, [id])

    def delete_many(self, ids):
        env = self._conn.open()
        self._conn.delete(env, ids)


LMDBDatabase.contractor_cls = LMDBCollection


class LMDBCursor(AbstractCursor):
    """
    """

    def __init__(self, collection, subject):
        super(LMDBCursor, self).__init__(collection, subject)
        self._conn = self._collection._conn
        self._env = self._conn.open()

    def query(self, max_scan):
        docs = (self._decode_doc(doc)
                for doc in self._conn.iter_docs(self._env))

        if not max_scan:
            return docs
        else:
            return islice(docs, max_scan)


LMDBCollection.contractor_cls = LMDBCursor

1
2	import os	×
3	import lmdb	×
4	import shutil	×
5	from itertools import islice	×
6
7	from ..types import unicode_	×
8	from . import (	×
9	AbstractStorage,
10	AbstractDatabase,
11	AbstractCollection,
12	AbstractCursor,
13
14	StorageDuplicateKeyError,
15	)
16
17	# Functional > Performence > Natural
18
19	LMDB_DB_EXT = ".mdb"	×
20
21
22	def _to_bytes(s):	×
23	return s.encode()	×
24
25
26	class LMDBKVEngine(object):	×
27	"""Per collection"""
28
29	dbname = _to_bytes("documents")	×
30
31	def __init__(self, options=None):	×
32	"""
33	"""
34	opt = options or dict()	×
35	opt.update({	×
36	"subdir": False,
37	"max_dbs": 1,
38	})
39	self.opt = opt	×
40	self._path = None	×
41
42	def set_path(self, path):	×
43	self._path = path	×
44
45	def open(self):	×
46	environment = lmdb.open(self._path, **self.opt)	×
47	return environment	×
48
49	def iter_docs(self, environment):	×
50	if not os.path.isfile(self._path):	×
51	return	×
52
53	with environment as env:	×
54	db = env.open_db(self.dbname)	×
55	with env.begin(db, write=False) as txn:	×
56	cursor = txn.cursor()	×
57	for encoded_doc in cursor.iternext(keys=False, values=True):	×
58	yield encoded_doc	×
59
60	def write(self, environment, pairs, overwrite=False):	×
61	if not os.path.isfile(self._path):	×
62	return	×
63
64	dup = False	×
65	with environment as env:	×
66	db = env.open_db(self.dbname)	×
67	with env.begin(db, write=True) as txn:	×
68	for doc_id, encoded_doc in pairs:	×
69	id = _to_bytes(str(doc_id))	×
70	if not txn.put(id, encoded_doc, overwrite=overwrite):	×
71	dup = True	×
72	break	×
73	if dup:	×
74	raise StorageDuplicateKeyError()	×
75
76	def delete(self, environment, doc_ids):	×
77	if not os.path.isfile(self._path):	×
78	return	×
79
80	with environment as env:	×
81	db = env.open_db(self.dbname, reverse_key=False)	×
82	with env.begin(db, write=True) as txn:	×
83	cursor = txn.cursor()	×
84	for doc_id in doc_ids:	×
85	id = _to_bytes(str(doc_id))	×
86	if cursor.set_key(id):	×
87	cursor.delete()	×
88
89
90	class LMDBStorage(AbstractStorage):	×
91	"""
92	"""
93
94	def __init__(self, repository, storage_config):	×
95	super(LMDBStorage, self).__init__(repository, storage_config)	×
96	self._conn = LMDBKVEngine(self._config)	×
97
98	def _db_path(self, db_name):	×
99	"""
100	Get Monty database dir path.
101	"""
102	return os.path.join(self._repository, db_name)	×
103
104	@classmethod	×
105	def nice_name(cls):
106	return "lightning"	×
107
108	@classmethod	×
109	def config(cls, map_size=10485760, **kwargs):	×
110	"""
111	"""
112	return {	×
113	"map_size": int(map_size),
114	}
115
116	def database_create(self, db_name):	×
117	if not os.path.isdir(self._db_path(db_name)):	×
118	os.makedirs(self._db_path(db_name))	×
119
120	def database_drop(self, db_name):	×
121	db_path = self._db_path(db_name)	×
122	if os.path.isdir(db_path):	×
123	shutil.rmtree(db_path)	×
124
125	def database_list(self):	×
126	return [	×
127	name for name in os.listdir(unicode_(self._repository))
128	if os.path.isdir(self._db_path(name))
129	]
130
131
132	class LMDBDatabase(AbstractDatabase):	×
133	"""
134	"""
135
136	def __init__(self, storage, subject):	×
137	super(LMDBDatabase, self).__init__(storage, subject)	×
138	self._db_path = storage._db_path(self._name)	×
139	self._conn = storage._conn	×
140
141	def _col_path(self, col_name):	×
142	"""
143	Get database file path, which is Monty collection.
144	"""
145	return os.path.join(self._db_path, col_name) + LMDB_DB_EXT	×
146
147	def database_exists(self):	×
148	return os.path.isdir(self._db_path)	×
149
150	def collection_exists(self, col_name):	×
151	return os.path.isfile(self._col_path(col_name))	×
152
153	def collection_create(self, col_name):	×
154	if not self.database_exists():	×
155	self._storage.database_create(self._name)	×
156	self._conn.set_path(self._col_path(col_name))	×
157	environment = self._conn.open()	×
158
159	return environment	×
160
161	def collection_drop(self, col_name):	×
162	if self.collection_exists(col_name):	×
163	os.remove(self._col_path(col_name))	×
164
165	def collection_list(self):	×
166	if not self.database_exists():	×
167	return []	×
168	return [os.path.splitext(name)[0]	×
169	for name in os.listdir(unicode_(self._db_path))
170	if name.endswith(LMDB_DB_EXT)]
171
172
173	LMDBStorage.contractor_cls = LMDBDatabase	×
174
175
176	class LMDBCollection(AbstractCollection):	×
177	"""
178	"""
179
180	def __init__(self, database, subject):	×
181	super(LMDBCollection, self).__init__(database, subject)	×
182	self._conn = database._conn	×
183	self._conn.set_path(database._col_path(self._name))	×
184
185	def _ensure_table(func):	×
186	def make_table(self, args, *kwargs):	×
187	environment = None	×
188	if not self._database.collection_exists(self._name):	×
189	environment = self._database.collection_create(self._name)	×
190	return func(self, env=environment, args, *kwargs)	×
191	return make_table	×
192
193	@_ensure_table	×
194	def write_one(self, doc, check_keys=True, env=None):	×
195	env = env or self._conn.open()	×
196
197	id = doc["_id"]	×
198	encoded = self._encode_doc(doc, check_keys)	×
199	self._conn.write(env, [(id, encoded)])	×
200
201	return id	×
202
203	@_ensure_table	×
204	def write_many(self, docs, check_keys=True, ordered=True, env=None):	×
205	env = env or self._conn.open()	×
206	ids = list()	×
207
208	def produce_encoded_docs():	×
209	for doc in docs:	×
210	id = doc["_id"]	×
211	yield id, self._encode_doc(doc, check_keys)	×
212	ids.append(id)	×
213
214	self._conn.write(env, produce_encoded_docs())	×
215
216	return ids	×
217
218	def update_one(self, doc):	×
219	env = self._conn.open()	×
220
221	id = doc["_id"]	×
222	encoded = self._encode_doc(doc)	×
223	self._conn.write(env, [(id, encoded)], overwrite=True)	×
224
225	def update_many(self, docs):	×
226	env = self._conn.open()	×
227
228	def produce_encoded_docs():	×
229	for doc in docs:	×
230	yield doc["_id"], self._encode_doc(doc)	×
231
232	self._conn.write(env, produce_encoded_docs(), overwrite=True)	×
233
234	def delete_one(self, id):	×
235	env = self._conn.open()	×
236	self._conn.delete(env, [id])	×
237
238	def delete_many(self, ids):	×
239	env = self._conn.open()	×
240	self._conn.delete(env, ids)	×
241
242
243	LMDBDatabase.contractor_cls = LMDBCollection	×
244
245
246	class LMDBCursor(AbstractCursor):	×
247	"""
248	"""
249
250	def __init__(self, collection, subject):	×
251	super(LMDBCursor, self).__init__(collection, subject)	×
252	self._conn = self._collection._conn	×
253	self._env = self._conn.open()	×
254
255	def query(self, max_scan):	×
256	docs = (self._decode_doc(doc)	×
257	for doc in self._conn.iter_docs(self._env))
258
259	if not max_scan:	×
260	return docs	×
261	else:
262	return islice(docs, max_scan)	×
263
264
265	LMDBCollection.contractor_cls = LMDBCursor	×

davidlatwe / montydb / 211

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous