Skip to content

Instantly share code, notes, and snippets.

@pfmoore
Created September 21, 2021 15:40
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save pfmoore/db41e1093c5dcc33b0e68fc8748eb2f9 to your computer and use it in GitHub Desktop.
Save pfmoore/db41e1093c5dcc33b0e68fc8748eb2f9 to your computer and use it in GitHub Desktop.
SQLite-backed persistent dictionary
import collections.abc
import sqlite3
class PersistentDict(collections.abc.MutableMapping):
def __init__(self, db=":memory:", table="dict"):
# Set to autocommit mode (isolation_level=None)
self.conn = sqlite3.connect(db, isolation_level=None)
if not table.isidentifier():
raise ValueError(f"Table name {table!r} is not a valid identifier")
self.table = table
self._create_table()
def _create_table(self):
sql = f"CREATE TABLE IF NOT EXISTS {self.table} (key PRIMARY KEY, value)"
self.conn.execute(sql)
def __del__(self):
if self.conn:
self.conn.close()
self.conn = None
def __getitem__(self, key, default=None):
sql = f"SELECT value FROM {self.table} WHERE key = ?"
# TODO: Unpacking error if key not found
vals = self.conn.execute(sql, (key,)).fetchone()
if vals is None:
raise KeyError(key)
return vals[0]
def __setitem__(self, key, value):
sql = f"INSERT INTO {self.table} (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value"
self.conn.execute(sql, (key, value))
def __delitem__(self, key):
sql = f"DELETE FROM {self.table} WHERE key = ?"
c = self.conn.execute(sql, (key,))
if c.rowcount == 0:
raise KeyError(key)
def __iter__(self):
sql = f"SELECT key FROM {self.table}"
for key, in self.conn.execute(sql):
yield key
def __len__(self):
sql = f"SELECT COUNT(*) FROM {self.table}"
n = self.conn.execute(sql).fetchone()
return n
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment