feat(app): make delete logs job an asynchronous scheduled task

api/main.py

@@ -96,7 +96,6 @@ async def lifespan(app: FastAPI):
     # Clean up resources, close connections, etc.
     await core_logger.shutdown()
     await async_core.shutdown()
-    source_collector_core.shutdown()
     await session.close()
     pass
 

collector_db/AsyncDatabaseClient.py

@@ -1,8 +1,9 @@
+from datetime import datetime, timedelta
 from functools import wraps
 from typing import Optional, Type, Any, List
 
 from fastapi import HTTPException
-from sqlalchemy import select, exists, func, case, desc, Select, not_, and_, or_, update, Delete, Insert, asc
+from sqlalchemy import select, exists, func, case, desc, Select, not_, and_, or_, update, Delete, Insert, asc, delete
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 from sqlalchemy.orm import selectinload, joinedload, QueryableAttribute, aliased
@@ -1615,3 +1616,15 @@
         logs = raw_results.scalars().all()
         return ([LogOutputInfo(**log.__dict__) for log in logs])
 
+    @session_manager
+    async def delete_old_logs(self, session):
+        """
+        Delete logs older than a day
+        """
+        statement = delete(Log).where(
+            Log.created_at < datetime.now() - timedelta(days=1)
+        )
+        await session.execute(statement)
+
+
+

collector_db/DatabaseClient.py

@@ -158,15 +158,6 @@ def get_batch_status(self, session, batch_id: int) -> BatchStatus:
         batch = session.query(Batch).filter_by(id=batch_id).first()
         return BatchStatus(batch.status)
 
-    @session_manager
-    def delete_old_logs(self, session):
-        """
-        Delete logs older than a day
-        """
-        session.query(Log).filter(
-            Log.created_at < datetime.now() - timedelta(days=1)
-        ).delete()
-
     @session_manager
     def update_url(self, session, url_info: URLInfo):
         url = session.query(URL).filter_by(id=url_info.id).first()

core/ScheduledTaskManager.py

@@ -1,41 +1,9 @@
 from datetime import datetime, timedelta
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
-from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.interval import IntervalTrigger
-
-from collector_db.DatabaseClient import DatabaseClient
 from core.AsyncCore import AsyncCore
 
-
-class ScheduledTaskManager:
-
-    def __init__(self, db_client: DatabaseClient):
-        # Dependencies
-        self.db_client = db_client
-
-        # Main objects
-        self.scheduler = BackgroundScheduler()
-        self.scheduler.start()
-        self.add_scheduled_tasks()
-
-        # Jobs
-        self.delete_old_logs_job = None
-
-
-    def add_scheduled_tasks(self):
-        self.delete_old_logs_job = self.scheduler.add_job(
-            self.db_client.delete_old_logs,
-            trigger=IntervalTrigger(
-                days=1,
-                start_date=datetime.now() + timedelta(minutes=10)
-            )
-        )
-
-    def shutdown(self):
-        if self.scheduler.running:
-            self.scheduler.shutdown()
-
 class AsyncScheduledTaskManager:
 
     def __init__(self, async_core: AsyncCore):
@@ -49,6 +17,7 @@ def __init__(self, async_core: AsyncCore):
 
         # Jobs
         self.run_cycles_job = None
+        self.delete_logs_job = None
 
     def add_scheduled_tasks(self):
         self.run_cycles_job = self.scheduler.add_job(
@@ -59,6 +28,13 @@ def add_scheduled_tasks(self):
             ),
             misfire_grace_time=60
         )
+        self.delete_logs_job = self.scheduler.add_job(
+            self.async_core.adb_client.delete_old_logs,
+            trigger=IntervalTrigger(
+                days=1,
+                start_date=datetime.now() + timedelta(minutes=10)
+            )
+        )
 
     def shutdown(self):
         if self.scheduler.running:

core/SourceCollectorCore.py

@@ -2,38 +2,17 @@
 
 
 from collector_db.DatabaseClient import DatabaseClient
-from core.ScheduledTaskManager import ScheduledTaskManager
 from core.enums import BatchStatus
 
 
 class SourceCollectorCore:
     def __init__(
         self,
         db_client: Optional[DatabaseClient] = None,
-        dev_mode: bool = False
     ):
         if db_client is None:
             db_client = DatabaseClient()
         self.db_client = db_client
-        if not dev_mode:
-            self.scheduled_task_manager = ScheduledTaskManager(db_client=db_client)
-        else:
-            self.scheduled_task_manager = None
-
 
     def get_status(self, batch_id: int) -> BatchStatus:
         return self.db_client.get_batch_status(batch_id)
-
-
-    def shutdown(self):
-        if self.scheduled_task_manager is not None:
-            self.scheduled_task_manager.shutdown()
-
-
-
-
-
-"""
-TODO: Add logic for batch processing
-
-"""

tests/test_automated/integration/collector_db/test_db_client.py

@@ -94,7 +94,7 @@ async def test_delete_old_logs(db_data_creator: DBDataCreator):
     db_client.insert_logs(log_infos=log_infos)
     logs = await adb_client.get_logs_by_batch_id(batch_id=batch_id)
     assert len(logs) == 3
-    db_client.delete_old_logs()
+    await adb_client.delete_old_logs()
 
     logs = await adb_client.get_logs_by_batch_id(batch_id=batch_id)
     assert len(logs) == 0

tests/test_automated/integration/conftest.py

@@ -13,10 +13,8 @@
 def test_core(db_client_test):
     core = SourceCollectorCore(
         db_client=db_client_test,
-        dev_mode=True
     )
     yield core
-    core.shutdown()
 
 
 @pytest.fixture