Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import collections.abc
- import sqlite3
- from contextlib import suppress
- from operator import itemgetter
- class SQLDict(collections.abc.MutableMapping):
- def __init__(self, dbname, items=[], **kwargs) -> None:
- self.dbname = dbname
- self.conn = sqlite3.connect(dbname)
- cursor = self.conn.cursor()
- with suppress(sqlite3.OperationalError):
- cursor.execute("CREATE TABLE Dict (key TEXT, value TEXT)")
- cursor.execute("CREATE UNIQUE INDEX Kndx ON Dict (key)")
- self.update(items, **kwargs)
- def __setitem__(self, key: str, value: str) -> None:
- if key in self:
- del self[key]
- with self.conn as c:
- c.execute("INSERT INTO Dict VALUES(?, ?)", (key, value))
- def __getitem__(self, key: str) -> str:
- c = self.conn.execute("SELECT value FROM Dict WHERE key=?", (key,))
- row = c.fetchone()
- if row is None:
- raise KeyError(key)
- return row[0]
- def __delitem__(self, key: str) -> None:
- if key not in self:
- raise KeyError(key)
- with self.conn as c:
- c.execute("DELETE FROM Dict WHERE key=?", (key,))
- def __len__(self) -> int:
- return next(self.conn.execute("SELECT COUNT(*) FROM Dict"))[0]
- def __iter__(self):
- c = self.conn.execute("SELECT key FROM Dict")
- return map(itemgetter(0), c.fetchall())
- def __repr__(self) -> str:
- return f"{type(self).__name__}(dbname={self.dbname!r}, items={list(self.items())})"
- def close(self):
- self.conn.close()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement