Das deutsche Python-Forum

Code: Alles auswählen

#!/usr/bin/env python3



import sqlite3

import pickle

import collections



class SqlDict(collections.MutableMapping):

	def __init__(self, db_path=':memory:',

					will_pickle=True,

					debug=False):

		

		if will_pickle:

			will_pickle = lambda i: not isinstance(i, (str, int, float, bool, None.__class__))

			

		self._sql_conn = conn = sqlite3.connect(db_path)

		self.will_pickle = will_pickle

		self.debug = debug

		self._execute('''

		CREATE TABLE IF NOT EXISTS dictitem

		(key BLOB,

		value BLOB,

		PRIMARY KEY(key))''')

	

	

	def __getitem__(self, key):

		fetched = self._execute('select value from dictitem where key=?', [key]).fetchone()

		if fetched is None:

			raise ValueError('key "{0}" not found'.format(key))

		return self._load(fetched[0])

		

	def __setitem__(self, key, value):

		self._execute('insert or replace into dictitem values (?, ?)', [key, value])

		

	def __delitem__(self, key):

		self[key]

		self._execute('delete from dictitem where key=?', [key])

		

	def __len__(self):

		return self._execute('select count(*) from dictitem').fetchone()[0]

		

	def __iter__(self):

		cursor = self._execute('select key from dictitem')

		while True:

			fetched = cursor.fetchone()

			if not fetched:

				break

			yield fetched[0]

			

	def close(self):

		self.sync()

		self._sql_conn.close()

	

	def sync(self):

		self._sql_conn.commit()

	

	def _load(self, arg):

		if isinstance(arg, bytes):

			return pickle.loads(arg)

		return arg

	

	def _dump(self, arg):

		if self.will_pickle and self.will_pickle(arg):

			return pickle.dumps(arg)

		return arg

	

	def _execute(self, query, args=[]):

		cursor = self._sql_conn.cursor()

		if self.will_pickle:

			args = list(map(self._dump, args))

		if self.debug:

			print('SQL executing:', query, args)

		cursor.execute(query, args)

		return cursor





if __name__ == '__main__':

	

	d = SqlDict()

	d['bla'] = range(8)

	

	import shelve

	s = shelve.Shelf(SqlDict('/tmp/test', will_pickle=False, debug=True), writeback=True)

	s['hype hyper'] = list(d.keys())

	s.update(d)

	

	s['bla'] = list(s['bla'])

	s['bla'].append('write back works!')

	print(s['bla'][-1])

	

	

	for key in s:

		print(key)

	

	

	d.close()

	s.close()

Code: Alles auswählen

normaler dict
fertig in 0.00016188621521

SqlDict
fertig in 0.065728187561

shelve mit dumbdbm
fertig in 2.27868795395

shelve mit gdbm
fertig in 2.36116504669

Code: Alles auswählen

#!/usr/bin/env python3

#import cProfile
from time import time
from sqldict import SqlDict
import shelve
import dbm

X = 500
g =  dict(zip(('key'+str(i) for i in range(X)), ['value']*X))

def test(d):
	t = time()
	
	d.update(g)
	len(d)
	for key in list(d.keys()):
		d[key]
		del d[key]
	
	print('fertig in '+ str(time()-t))
	print()
		

print('normaler dict')
test({})

print ('SqlDict')
test(SqlDict('/tmp/foobadfdfdsar'))

print ('shelve mit dumbdbm')
dbm._names = ['dbm.dumb']
test(shelve.Shelf(dbm.open('/tmp/teafdssdfadsljkla',  'n')))

print('shelve mit gdbm')
dbm._names = ['dbm.gnu']
test(shelve.open('/tmp/fooasaadfrbaz'))

Code: Alles auswählen

def __iter__(self):
    with self._execute(...) as cursor:
        for row in cursor:
            yield row[0]

Code: Alles auswählen

normaler dict
fertig in 0.000365018844604

SqlDict
fertig in 9.7190990448

shelve mit dumbdbm
fertig in 2.42650508881