Add sqlite3/db.py

sqlite3/db.py

@@ -0,0 +1,52 @@
+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())