Update DB logic

core/medium_parser/__init__.py

@@ -1,5 +1,6 @@
 
 from aiohttp_retry import ExponentialRetry
+from loguru import logger
 
 import jinja2
 from database_lib import SQLiteCacheBackend
@@ -8,6 +9,8 @@ cache = SQLiteCacheBackend('medium_db_cache.sqlite')
 cache.init_db()
 cache.enable_zstd()
 
+logger.debug(f"Database length: {cache.all_length()}")
+
 retry_options = ExponentialRetry(attempts=3)
 
 from . import exceptions as exceptions

core/medium_parser/core.py

@@ -86,9 +86,11 @@ class MediumParser:
 
     async def query(self, use_cache: bool = True, retry: int = 3):
         logger.debug(f"Medium QUERY: {use_cache=}, {retry=}")
+        cache_used = True
         post_data = await self.get_post_data_from_cache() if use_cache else None
 
         if not post_data:
+            cache_used = False
             logger.debug("Getting value from cache failed, using API")
 
         attempt = 0
@@ -104,10 +106,24 @@ class MediumParser:
                 await asyncio.sleep(2 ** attempt)
                 attempt += 1
 
-        if not post_data or not isinstance(post_data, dict) or post_data.get("error") or not post_data.get("data") or not post_data.get("data").get("post"):
-            raise MediumPostQueryError(f'Could not query post by ID from API: {self.post_id}')
+        reason = None
 
-        cache.push(self.post_id, post_data)
+        if not post_data:
+            reason = "No post data returned"
+        elif not isinstance(post_data, dict):
+            reason = "Post data is not a dictionary"
+        elif post_data.get("error"):
+            reason = "Post data contains an error"
+        elif not post_data.get("data"):
+            reason = "Post data missing 'data' key"
+        elif not post_data.get("data").get("post"):
+            reason = "Post data missing 'data.post' key"
+
+        if reason:
+            raise MediumPostQueryError(f'Could not query post by ID from API: {self.post_id}. Reason: {reason}')
+
+        if not cache_used:
+            cache.push(self.post_id, post_data)
 
         self.post_data = post_data
         return self.post_data

database-lib/database_lib/cache_db.py

@@ -3,6 +3,7 @@ import sqlite3
 import json
 from loguru import logger
 from warnings import warn
+import threading
 try:
     import sqlite_zstd
 except ImportError:
@@ -59,8 +60,7 @@ class SQLiteCacheBackend:
             except Exception as error:
                 print(error)
             self.connection.execute("PRAGMA auto_vacuum=full")
-            self.cursor.execute("SELECT zstd_incremental_maintenance(null, 1);")
-            self.cursor.execute("vacuum;")
+            self.maintenance_thread()
 
     def init_db(self):
         with self.connection:
@@ -97,21 +97,30 @@ class SQLiteCacheBackend:
             else:
                 logger.debug(f"Attempted to delete non-existing key: {key}")
     
-    def maintenance(self):
+    def maintenance(self, time: int = None, blocking_time: int = 0.5):
         with self.connection:
+            if time is not None:
+                self.cursor.execute("SELECT zstd_incremental_maintenance(?, ?);", (time, blocking_time))
+            else:
+                self.cursor.execute("SELECT zstd_incremental_maintenance(null, ?);", (blocking_time,))
             self.cursor.execute("VACUUM")
             self.cursor.execute("ANALYZE")
 
-    def migrate_add_index_to_key(self):
-        with self.connection:
-            # Check if the index already exists
-            index_exists = self.cursor.execute("SELECT name FROM sqlite_master WHERE type='index' AND name='idx_key'").fetchone()
-            if not index_exists:
-                # Create the index if it doesn't exist
-                self.cursor.execute("CREATE INDEX idx_key ON cache (key)")
-                logger.info("Index 'idx_key' on column 'key' created successfully.")
-            else:
-                logger.info("Index 'idx_key' on column 'key' already exists.")
+    def maintenance_thread(self):
+        maintenance_thread = threading.Thread(target=self.maintenance, daemon=True)
+        maintenance_thread.start()
+
+    # Doesn't works with sqlite_zstd
+    # def migrate_add_index_to_key(self):
+    #     with self.connection:
+    #         # Check if the index already exists
+    #         index_exists = self.cursor.execute("SELECT name FROM sqlite_master WHERE type='index' AND name='idx_key'").fetchone()
+    #         if not index_exists:
+    #             # Create the index if it doesn't exist
+    #             self.cursor.execute("CREATE INDEX idx_key ON cache (key)")
+    #             logger.info("Index 'idx_key' on column 'key' created successfully.")
+    #         else:
+    #             logger.info("Index 'idx_key' on column 'key' already exists.")
 
     def show_schema_info(self):
         with self.connection: