github.com/hoffmann Peter Hoffmann on Stackoverflow @peterhoffmann on twitter Peter Hoffmann on Facebook Contact me per email Subscribe to Atom Feed

Peter Hoffmann

Software Engineer
prev page next page

Adding a dict based interface to the python leveldb api

Posted on August 6, 2011
#python #nosql

In the last post I showed how to install the python LevelDB bindings. Here is how to wrap the python LevelDB API into a dict like API with the help of UserDict.DictMixin. In a subclass we will extend the the simple key<string>:value<string> mapping to a dict with arbitrary JSON serializablie values.

from UserDict import DictMixin
import leveldb
import json


class LevelDict(DictMixin):
    """Dict Wrapper around the Google LevelDB Database"""
    def init(self, path):
        """Constructor for LevelDict"""
        self.path = path
        self.db = leveldb.LevelDB(self.path)

    def getitem(self, key):
        return self.db.Get(key)

    def setitem(self, key, value):
        self.db.Put(key, value)

    def delitem(self, key):
        self.db.Delete(key)

    def iter(self):
        for k, v in self.db.RangeIter():
            yield v

    def keys(self):
        return [k for k, v in self.db.RangeIter()]

    def iteritems(self):
        return self.db.RangeIter()

    def rangescan(self, start=None, end=None):
        if start is None and end is None:
            return self.db.RangeIter()
        elif end is None:
            return self.db.RangeIter(start)
        else:
            return self.db.RangeIter(start, end)

To store more than just strings we will add JSON serialization to a subclass:

class LevelJsonDict(LevelDict):
    """Dict Wrapper around the Google LevelDB Database with JSON serialization"""
    def __getitem__(self, key):
        return json.loads(LevelDict.__getitem__(self, key))

    def __setitem__(self, key, value):
        LevelDict.__setitem__(self, key, json.dumps(value))

    def iteritems(self):
        for k, v in LevelDict.iteritems(self):
            yield k, json.loads(v)

    def rangescan(self, start=None, end=None):
        for k, v in LevelDict.rangescan(self, start, end):
            yield k, json.loads(v)

Now you have a fast JSON Store with a Dict based Interface and a LevelDB backend.