52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
import sqlite3, json
|
|
|
|
class DB(sqlite3.Connection):
|
|
def __init__(self, db_name="db.sqlite"):
|
|
super().__init__(db_name)
|
|
with self:
|
|
self.execute('''
|
|
CREATE TABLE IF NOT EXISTS kv_store
|
|
(key TEXT PRIMARY KEY, value BLOB)
|
|
''')
|
|
|
|
def __setitem__(self, key, value):
|
|
if not isinstance(value, bytes): value = json.dumps(value)
|
|
with self:
|
|
cur = self.execute('''
|
|
INSERT OR REPLACE INTO kv_store
|
|
(key, value) VALUES (?, ?)
|
|
''', (key, value))
|
|
return {"modified_count": cur.rowcount}
|
|
|
|
def __getitem__(self, key):
|
|
with self:
|
|
result = self.execute('''
|
|
SELECT value FROM kv_store
|
|
WHERE key = ?
|
|
''', (key,)).fetchone()
|
|
if result:
|
|
if isinstance(result[0], str):
|
|
try: return json.loads(result[0])
|
|
except json.JSONDecodeError: pass
|
|
return result[0]
|
|
|
|
def delete(self, key):
|
|
with self:
|
|
cur = self.execute('''
|
|
DELETE FROM kv_store
|
|
WHERE key = ?
|
|
''', (key,))
|
|
return {"deleted_count": cur.rowcount}
|
|
|
|
def keys(self, pattern='*'):
|
|
pattern = pattern.translate(str.maketrans({
|
|
'\\': '\\\\', '%': '\\%', '_': '\\_', '*': '%', '?': '_'
|
|
}))
|
|
with self:
|
|
result = self.execute('''
|
|
SELECT key FROM kv_store
|
|
WHERE key LIKE ? ESCAPE '\\'
|
|
''', (pattern,)).fetchall()
|
|
return [row[0] for row in result]
|
|
|
|
def __repr__(self): return repr(self.keys()) |