add initial review testing for sql

src/asquare.py

@@ -4,6 +4,12 @@ import json
 import asyncio
 import uuid
 import re
+from time import time
+import re
+# we dont need no stinkin floats
+time = int(time())
+
+import db
 
 import get_sq_env
 
@@ -114,15 +120,37 @@ async def acat2id(cat_id: str):
 
     return items
 
+async def do_review() -> bool:
+    result = await acat2id(LEGO_CAT_ID)
+    for item in result:
+        if not re.match(r".*\d{5}.*", item["name"]):
+            missing = []
+            print(f"{item} is missing its set number!")
+            print(f"[@DEBUG] Upserting {item['id']} into REVIEW table")
+            item_sanitized = {
+                "sq_object": item["id"],
+                "set_name": item["name"],
+                "set_number": None,
+                "time_added": time
+            }
+            missing.append(item_sanitized)
+    for item in missing:
+        sq_item = await db.fetch(item["id"])  #
+        if item["id"] != sq_item["sq_object"]:
+            await db.upsert("REVIEW", item_sanitized, "sq_object")
+            print(f"[@DEBUG] Upserted {item['id']} into REVIEW table")
+        print(f"[@DEBUG] {item['id']} already in REVIEW table")
+    return True
+
 
 async def test() -> None:
     task = asyncio.create_task(acat2id(LEGO_CAT_ID))  # type: ignore
+    await do_review()
     await asyncio.sleep(1)  # Give the task some time to complete
     result = await task
-    print("OBJECTS IN LEGO CATEGORY:")
-    for item in result:
-        print(f"ID: {item['id']}, Name: {item['name']}")
-    # eturn
+    #print("OBJECTS IN LEGO CATEGORY:")
+        #print(f"ID: {item['id']}, Name: {item['name']}")
+    return True
     # pass
 
 
@@ -134,6 +162,7 @@ async def test() -> None:
 
 
 if __name__ == "__main__":
+    print("Spooling up..")
     print(f"@[DEBUG] ENV:\t{ENVIRONMENT}")
     print(f"@[DEBUG] CATID:\t{LEGO_CAT_ID}")
     asyncio.run(test())

src/bricksetapi.py

@@ -33,9 +33,9 @@ def get_set(set_number: int) -> dict:
     response = requests.get(url)
     if response.status_code == 200:
         data = response.json()
-        if not data["sets"] and if data["message"] == "API limit exceeded":
+        if not data["sets"] and data["message"] == "API limit exceeded":
             raise Exception("API limit exceeded. Please try again later.")
-            break
+            return False
 
         print(f"[@DEBUG]\t{data}")
         our_set = {

src/db.py

@@ -0,0 +1,53 @@
+import aiosqlite
+from time import time
+time = int(time())  # we dont need no stinkin floats
+
+# db = await aiosqlite.connect("edderkop.db")  # type: ignore
+async def fetch(sqobj):
+    """
+    Asynchronously fetch a record from the REVIEW table by sq_object.
+
+    Args:
+        sqobj (str): The sq_object identifier to search for.
+    Returns:
+        dict or None: The fetched record as a dictionary, or None if not found.
+    """
+    async with aiosqlite.connect("edderkop.db") as db:  # type: ignore
+        db.row_factory = aiosqlite.Row  # type: ignore
+        async with db.execute(
+            "SELECT * FROM REVIEW WHERE sq_object = ?", (sqobj,)
+        ) as cursor:
+            row = await cursor.fetchone()
+            if row:
+                return dict(row)
+            return None
+
+async def upsert(table: str, data: dict, unique_key: str) -> None:
+    """
+    Asynchronously insert or update a record in the specified table.
+
+    Args:
+        sq_ib (aiosqlite.Connection): The SQLite database connection.
+        table (str): The name of the table to upsert into.
+        data (dict): A dictionary of column-value pairs to insert or update.
+        unique_key (str): The column name that serves as the unique key for upsert.
+
+    Returns:
+        None
+    """
+    print(f"Upserting into {table}: {data}")
+
+    columns = ', '.join(data.keys())
+    placeholders = ', '.join('?' * len(data))
+    updates = ', '.join(f"{col}=excluded.{col}" for col in data.keys() if col != unique_key)
+
+    sql = f"""
+    INSERT INTO {table} ({columns})
+    VALUES ({placeholders})
+    ON CONFLICT({unique_key}) DO UPDATE SET
+    {updates};
+    """
+    async with aiosqlite.connect("edderkop.db") as db:  # type: ignore
+        async with db.execute(sql, tuple(data.values())) as cursor:
+            await db.commit()
+            #await db.close()

src/scheduler.py

@@ -1,221 +1,221 @@
-import aiosqlite
-import logging
-from typing import Optional, List, Dict, Any
-import json
-import os
-
-logger = logging.getLogger(__name__)
-
-# Get the absolute path to edderkop.db in the project root
-DEFAULT_DB_PATH = os.path.join(
-    os.path.dirname(os.path.dirname(__file__)), "edderkop.db"
-)
-
-if not os.path.exists(DEFAULT_DB_PATH):
-    raise FileNotFoundError(f"Database file not found at {DEFAULT_DB_PATH}")
-    print('Please run "src/scripts/EDDERKOP.SQL" to setup the database.')
-    os.exit(0)
-
-
-class AsyncSchedulerDB:
-    """
-    Asynchronous SQLite database manager for APScheduler.
-    Handles job storage, retrieval, and database operations using the main edderkop.db database.
-    """
-
-    def __init__(self, db_path: str = DEFAULT_DB_PATH):
-        """
-        Initialize the database manager.
-
-        Args:
-            db_path (str, optional): Path to the SQLite database file.
-                                   Defaults to edderkop.db in project root.
-        """
-        self.db_path = db_path
-        self._initialized = False
-        logger.info(f"Using database at: {self.db_path}")
-
-    async def initialize(self):
-        """Initialize the database schema if it doesn't exist."""
-        if self._initialized:
-            return
-
-        async with aiosqlite.connect(self.db_path) as db:
-            # Create apscheduler jobs table if it doesn't exist
-            await db.execute(
-                """
-                CREATE TABLE IF NOT EXISTS apscheduler_jobs (
-                    id VARCHAR(191) NOT NULL,
-                    next_run_time FLOAT,
-                    job_state BLOB NOT NULL,
-                    PRIMARY KEY (id)
-                )
-            """
-            )
-            await db.commit()
-
-        self._initialized = True
-        logger.info(f"Initialized scheduler database at {self.db_path}")
-
-    async def get_all_jobs(self) -> List[Dict[str, Any]]:
-        """
-        Retrieve all jobs from the database.
-
-        Returns:
-            List[Dict[str, Any]]: List of job dictionaries
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
-            async with db.execute(
-                "SELECT * FROM apscheduler_jobs ORDER BY next_run_time"
-            ) as cursor:
-                rows = await cursor.fetchall()
-                return [
-                    {
-                        "id": row["id"],
-                        "next_run_time": row["next_run_time"],
-                        "job_state": json.loads(row["job_state"]),
-                    }
-                    for row in rows
-                ]
-
-    async def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
-        """
-        Retrieve a specific job by ID.
-
-        Args:
-            job_id (str): The ID of the job to retrieve
-
-        Returns:
-            Optional[Dict[str, Any]]: Job dictionary if found, None otherwise
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
-            async with db.execute(
-                "SELECT * FROM apscheduler_jobs WHERE id = ?", (job_id,)
-            ) as cursor:
-                row = await cursor.fetchone()
-                if row:
-                    return {
-                        "id": row["id"],
-                        "next_run_time": row["next_run_time"],
-                        "job_state": json.loads(row["job_state"]),
-                    }
-                return None
-
-    async def add_job(
-        self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
-    ) -> bool:
-        """
-        Add a new job to the database.
-
-        Args:
-            job_id (str): Unique identifier for the job
-            next_run_time (float): Unix timestamp for next scheduled run
-            job_state (Dict[str, Any]): Job state dictionary
-
-        Returns:
-            bool: True if successful, False otherwise
-        """
-        await self.initialize()
-        try:
-            async with aiosqlite.connect(self.db_path) as db:
-                await db.execute(
-                    "INSERT INTO apscheduler_jobs (id, next_run_time, job_state) VALUES (?, ?, ?)",
-                    (job_id, next_run_time, json.dumps(job_state)),
-                )
-                await db.commit()
-                return True
-        except aiosqlite.IntegrityError:
-            logger.error(f"Job with ID {job_id} already exists")
-            return False
-        except Exception as e:
-            logger.error(f"Error adding job: {e}")
-            return False
-
-    async def update_job(
-        self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
-    ) -> bool:
-        """
-        Update an existing job in the database.
-
-        Args:
-            job_id (str): ID of the job to update
-            next_run_time (float): New next run time
-            job_state (Dict[str, Any]): Updated job state
-
-        Returns:
-            bool: True if successful, False if job not found
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            cursor = await db.execute(
-                """UPDATE apscheduler_jobs 
-                   SET next_run_time = ?, job_state = ?
-                   WHERE id = ?""",
-                (next_run_time, json.dumps(job_state), job_id),
-            )
-            await db.commit()
-            return cursor.rowcount > 0
-
-    async def remove_job(self, job_id: str) -> bool:
-        """
-        Remove a job from the database.
-
-        Args:
-            job_id (str): ID of the job to remove
-
-        Returns:
-            bool: True if job was removed, False if not found
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            cursor = await db.execute(
-                "DELETE FROM apscheduler_jobs WHERE id = ?", (job_id,)
-            )
-            await db.commit()
-            return cursor.rowcount > 0
-
-    async def remove_all_jobs(self) -> int:
-        """
-        Remove all jobs from the database.
-
-        Returns:
-            int: Number of jobs removed
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            cursor = await db.execute("DELETE FROM apscheduler_jobs")
-            await db.commit()
-            return cursor.rowcount
-
-    async def get_due_jobs(self, now: float) -> List[Dict[str, Any]]:
-        """
-        Get all jobs that are due to run.
-
-        Args:
-            now (float): Current timestamp
-
-        Returns:
-            List[Dict[str, Any]]: List of due jobs
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
-            async with db.execute(
-                """SELECT * FROM apscheduler_jobs 
-                   WHERE next_run_time <= ? 
-                   ORDER BY next_run_time""",
-                (now,),
-            ) as cursor:
-                rows = await cursor.fetchall()
-                return [
-                    {
-                        "id": row["id"],
-                        "next_run_time": row["next_run_time"],
-                        "job_state": json.loads(row["job_state"]),
-                    }
-                    for row in rows
-                ]
+import aiosqlite
+import logging
+from typing import Optional, List, Dict, Any
+import json
+import os
+
+logger = logging.getLogger(__name__)
+
+# Get the absolute path to edderkop.db in the project root
+DEFAULT_DB_PATH = os.path.join(
+    os.path.dirname(os.path.dirname(__file__)), "edderkop.db"
+)
+
+if not os.path.exists(DEFAULT_DB_PATH):
+    raise FileNotFoundError(f"Database file not found at {DEFAULT_DB_PATH}")
+    print('Please run "src/scripts/EDDERKOP.SQL" to setup the database.')
+    os.exit(0)
+
+
+class AsyncSchedulerDB:
+    """
+    Asynchronous SQLite database manager for APScheduler.
+    Handles job storage, retrieval, and database operations using the main edderkop.db database.
+    """
+
+    def __init__(self, db_path: str = DEFAULT_DB_PATH):
+        """
+        Initialize the database manager.
+
+        Args:
+            db_path (str, optional): Path to the SQLite database file.
+                                   Defaults to edderkop.db in project root.
+        """
+        self.db_path = db_path
+        self._initialized = False
+        logger.info(f"Using database at: {self.db_path}")
+
+    async def initialize(self):
+        """Initialize the database schema if it doesn't exist."""
+        if self._initialized:
+            return
+
+        async with aiosqlite.connect(self.db_path) as db:
+            # Create apscheduler jobs table if it doesn't exist
+            await db.execute(
+                """
+                CREATE TABLE IF NOT EXISTS apscheduler_jobs (
+                    id VARCHAR(191) NOT NULL,
+                    next_run_time FLOAT,
+                    job_state BLOB NOT NULL,
+                    PRIMARY KEY (id)
+                )
+            """
+            )
+            await db.commit()
+
+        self._initialized = True
+        logger.info(f"Initialized scheduler database at {self.db_path}")
+
+    async def get_all_jobs(self) -> List[Dict[str, Any]]:
+        """
+        Retrieve all jobs from the database.
+
+        Returns:
+            List[Dict[str, Any]]: List of job dictionaries
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                "SELECT * FROM apscheduler_jobs ORDER BY next_run_time"
+            ) as cursor:
+                rows = await cursor.fetchall()
+                return [
+                    {
+                        "id": row["id"],
+                        "next_run_time": row["next_run_time"],
+                        "job_state": json.loads(row["job_state"]),
+                    }
+                    for row in rows
+                ]
+
+    async def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
+        """
+        Retrieve a specific job by ID.
+
+        Args:
+            job_id (str): The ID of the job to retrieve
+
+        Returns:
+            Optional[Dict[str, Any]]: Job dictionary if found, None otherwise
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                "SELECT * FROM apscheduler_jobs WHERE id = ?", (job_id,)
+            ) as cursor:
+                row = await cursor.fetchone()
+                if row:
+                    return {
+                        "id": row["id"],
+                        "next_run_time": row["next_run_time"],
+                        "job_state": json.loads(row["job_state"]),
+                    }
+                return None
+
+    async def add_job(
+        self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
+    ) -> bool:
+        """
+        Add a new job to the database.
+
+        Args:
+            job_id (str): Unique identifier for the job
+            next_run_time (float): Unix timestamp for next scheduled run
+            job_state (Dict[str, Any]): Job state dictionary
+
+        Returns:
+            bool: True if successful, False otherwise
+        """
+        await self.initialize()
+        try:
+            async with aiosqlite.connect(self.db_path) as db:
+                await db.execute(
+                    "INSERT INTO apscheduler_jobs (id, next_run_time, job_state) VALUES (?, ?, ?)",
+                    (job_id, next_run_time, json.dumps(job_state)),
+                )
+                await db.commit()
+                return True
+        except aiosqlite.IntegrityError:
+            logger.error(f"Job with ID {job_id} already exists")
+            return False
+        except Exception as e:
+            logger.error(f"Error adding job: {e}")
+            return False
+
+    async def update_job(
+        self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
+    ) -> bool:
+        """
+        Update an existing job in the database.
+
+        Args:
+            job_id (str): ID of the job to update
+            next_run_time (float): New next run time
+            job_state (Dict[str, Any]): Updated job state
+
+        Returns:
+            bool: True if successful, False if job not found
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            cursor = await db.execute(
+                """UPDATE apscheduler_jobs 
+                   SET next_run_time = ?, job_state = ?
+                   WHERE id = ?""",
+                (next_run_time, json.dumps(job_state), job_id),
+            )
+            await db.commit()
+            return cursor.rowcount > 0
+
+    async def remove_job(self, job_id: str) -> bool:
+        """
+        Remove a job from the database.
+
+        Args:
+            job_id (str): ID of the job to remove
+
+        Returns:
+            bool: True if job was removed, False if not found
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            cursor = await db.execute(
+                "DELETE FROM apscheduler_jobs WHERE id = ?", (job_id,)
+            )
+            await db.commit()
+            return cursor.rowcount > 0
+
+    async def remove_all_jobs(self) -> int:
+        """
+        Remove all jobs from the database.
+
+        Returns:
+            int: Number of jobs removed
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            cursor = await db.execute("DELETE FROM apscheduler_jobs")
+            await db.commit()
+            return cursor.rowcount
+
+    async def get_due_jobs(self, now: float) -> List[Dict[str, Any]]:
+        """
+        Get all jobs that are due to run.
+
+        Args:
+            now (float): Current timestamp
+
+        Returns:
+            List[Dict[str, Any]]: List of due jobs
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                """SELECT * FROM apscheduler_jobs 
+                   WHERE next_run_time <= ? 
+                   ORDER BY next_run_time""",
+                (now,),
+            ) as cursor:
+                rows = await cursor.fetchall()
+                return [
+                    {
+                        "id": row["id"],
+                        "next_run_time": row["next_run_time"],
+                        "job_state": json.loads(row["job_state"]),
+                    }
+                    for row in rows
+                ]

src/scripts/throttler_example.py

@@ -1,15 +1,15 @@
-import aiohttp
-import asyncio
-
-from asyncio_throttle.throttler import Throttler
-
-throttler = Throttler(rate_limit=500, period=60)  # 500 requests per minute
-
-
-async def worker(throttler, session):
-    while True:
-        async with throttler:
-            async with session.get("http://example.com") as resp:
-                do_some_job_with(await resp.text())
-
-        await asyncio.sleep(0.05)
+import aiohttp
+import asyncio
+
+from asyncio_throttle.throttler import Throttler
+
+throttler = Throttler(rate_limit=500, period=60)  # 500 requests per minute
+
+
+async def worker(throttler, session):
+    while True:
+        async with throttler:
+            async with session.get("http://example.com") as resp:
+                do_some_job_with(await resp.text())
+
+        await asyncio.sleep(0.05)