Improved readability

openrecall/database.py

@@ -1,62 +1,46 @@
 import sqlite3
+from collections import namedtuple
+from typing import Any, List
 
 from openrecall.config import db_path
 
-
+Entry = namedtuple("Entry", ["id", "app", "title", "text", "timestamp", "embedding"])
-def create_db():
-    conn = sqlite3.connect(db_path)
-    c = conn.cursor()
-    c.execute(
-        """CREATE TABLE IF NOT EXISTS entries
-           (id INTEGER PRIMARY KEY AUTOINCREMENT, app TEXT, title TEXT, text TEXT, timestamp INTEGER, embedding BLOB)"""
-    )
-    conn.commit()
-    conn.close()
 
 
-def get_all_entries():
+def create_db() -> None:
-    conn = sqlite3.connect(db_path)
+    with sqlite3.connect(db_path) as conn:
-    c = conn.cursor()
+        c = conn.cursor()
-    results = c.execute("SELECT * FROM entries").fetchall()
+        c.execute(
-    conn.close()
+            """CREATE TABLE IF NOT EXISTS entries
-    entries = []
+               (id INTEGER PRIMARY KEY AUTOINCREMENT, app TEXT, title TEXT, text TEXT, timestamp INTEGER, embedding BLOB)"""
-    for result in results:
-        entries.append(
-            {
-                "id": result[0],
-                "app": result[1],
-                "title": result[2],
-                "text": result[3],
-                "timestamp": result[4],
-                "embedding": result[5],
-            }
         )
-    return entries
+        conn.commit()
 
 
-def get_timestamps():
+def get_all_entries() -> List[Entry]:
-    conn = sqlite3.connect(db_path)
+    with sqlite3.connect(db_path) as conn:
-    c = conn.cursor()
+        c = conn.cursor()
-    results = c.execute(
+        results = c.execute("SELECT * FROM entries").fetchall()
-        "SELECT timestamp FROM entries ORDER BY timestamp DESC LIMIT 1000"
+        return [Entry(*result) for result in results]
-    ).fetchall()
-    timestamps = [result[0] for result in results]
-    conn.close()
-    return timestamps
 
-def insert_entry(text, timestamp, embedding, app, title):
+
-    conn = sqlite3.connect(db_path)
+def get_timestamps() -> List[int]:
-    c = conn.cursor()
+    with sqlite3.connect(db_path) as conn:
+        c = conn.cursor()
+        results = c.execute(
+            "SELECT timestamp FROM entries ORDER BY timestamp DESC LIMIT 1000"
+        ).fetchall()
+        return [result[0] for result in results]
+
+
+def insert_entry(
+    text: str, timestamp: int, embedding: Any, app: str, title: str
+) -> None:
     embedding_bytes = embedding.tobytes()
-    c.execute(
+    with sqlite3.connect(db_path) as conn:
-        "INSERT INTO entries (text, timestamp, embedding, app, title) VALUES (?, ?, ?, ?, ?)",
+        c = conn.cursor()
-        (
+        c.execute(
-            text,
+            "INSERT INTO entries (text, timestamp, embedding, app, title) VALUES (?, ?, ?, ?, ?)",
-            timestamp,
+            (text, timestamp, embedding_bytes, app, title),
-            embedding_bytes,
+        )
-            app,
+        conn.commit()
-            title,
-        ),
-    )
-    conn.commit()
-    conn.close()