|
import os |
|
import time |
|
import shutil |
|
|
|
# The databases! |
|
from redis import Redis |
|
from unqlite import UnQLite |
|
from vedis import Vedis |
|
import semidbm.db as SemiDBM |
|
import sqlitedict |
|
import tinydb |
|
import pickledb |
|
import dbm.dumb |
|
|
|
|
|
|
|
def timed(fn): |
|
def inner(*args): |
|
s = time.time() |
|
fn(*args) |
|
return time.time() - s |
|
return inner |
|
|
|
|
|
class Test(object): |
|
def __init__(self, n): |
|
self.n = n |
|
if os.path.exists(self.filename): |
|
if os.path.isdir(self.filename): |
|
shutil.rmtree(self.filename, ignore_errors=True) |
|
else: |
|
os.unlink(self.filename) |
|
self._db = self.get_db() |
|
|
|
def close(self): |
|
try: |
|
self._db.close() |
|
except AttributeError: |
|
pass |
|
|
|
@timed |
|
def test_writes(self): |
|
for i in range(self.n): |
|
self.write(str(i)) |
|
|
|
@timed |
|
def test_reads(self): |
|
for i in range(self.n): |
|
self.read(str(i)) |
|
|
|
def write(self, i): |
|
self._db[i] = i |
|
|
|
def read(self, i): |
|
return self._db[i] |
|
|
|
|
|
class UnQLiteTest(Test): |
|
filename = '/tmp/unq.db' |
|
|
|
def get_db(self): |
|
return UnQLite(self.filename) |
|
|
|
|
|
class UnQLiteJSONTest(Test): |
|
filename = '/tmp/unq-json.db' |
|
|
|
def get_db(self): |
|
db = UnQLite(self.filename) |
|
self.coll = db.collection('values') |
|
self.coll.create() |
|
|
|
@timed |
|
def test_writes(self): |
|
data = [{str(i): i} for i in range(self.n)] |
|
self.coll.store(data) |
|
|
|
@timed |
|
def test_reads(self): |
|
return self.coll.all() |
|
|
|
|
|
class VedisTest(Test): |
|
filename = '/tmp/vedis.db' |
|
|
|
def get_db(self): |
|
return Vedis(self.filename) |
|
|
|
|
|
class SemidbmTest(Test): |
|
filename = '/tmp/semidbm.tmp' |
|
|
|
def get_db(self): |
|
return SemiDBM.open(self.filename, 'c') |
|
|
|
|
|
class SQLitedictTest(Test): |
|
filename = '/tmp/sqldict.db' |
|
|
|
def get_db(self): |
|
return sqlitedict.open(self.filename) |
|
|
|
|
|
class DBMdumbTest(Test): |
|
filename = '/tmp/dbmdumb.db' |
|
|
|
def get_db(self): |
|
return dbm.dumb.open(self.filename) |
|
|
|
|
|
class PickleDBTest(Test): |
|
filename = '/tmp/pickle.db' |
|
|
|
def get_db(self): |
|
return pickledb.load(self.filename, True) |
|
|
|
@timed |
|
def test_writes(self): |
|
for i in range(self.n): |
|
self.get_db().set(str(i), i) |
|
|
|
@timed |
|
def test_reads(self): |
|
return self._db.getall() |
|
|
|
|
|
class TinydbTest(Test): |
|
filename = '/tmp/tinydbm.json' |
|
|
|
def get_db(self): |
|
return tinydb.TinyDB(self.filename) |
|
|
|
@timed |
|
def test_writes(self): |
|
for i in range(self.n): |
|
self._db.insert({"key": str(i),"value": i}) |
|
|
|
@timed |
|
def test_reads(self): |
|
return self._db.storage.read() |
|
|
|
|
|
class RedisTest(Test): |
|
filename = '/tmp/redis' |
|
|
|
def get_db(self): |
|
db = Redis(db=15) |
|
db.flushdb() |
|
return db |
|
|
|
|
|
tests = [ |
|
UnQLiteTest, |
|
UnQLiteJSONTest, |
|
VedisTest, |
|
SemidbmTest, |
|
DBMdumbTest, |
|
SQLitedictTest, |
|
RedisTest, # Slow |
|
TinydbTest, # Extra Slow |
|
PickleDBTest, # Extra Slow |
|
] |
|
|
|
N = 10000 |
|
|
|
print('Testing with N =', N) |
|
print('------------------------------------') |
|
|
|
for TestClass in tests: |
|
test_name = TestClass.__name__.replace('Test', '') |
|
print(test_name) |
|
print('~' * len(test_name)) |
|
test = TestClass(N) |
|
print('Writes: ', test.test_writes()) |
|
print('Reads: ', test.test_reads()) |
|
print('') |
|
test.close() |