update db

app.py

@@ -63,7 +63,7 @@ async def lifespan(app: FastAPI):
     
     # Shutdown: Upload DB to Drive
     logger.info("Shutdown: Uploading database to Google Drive...")
-    drive_service.upload_db()
+    await drive_service.upload_db_async()
     logger.info("Shutting down...")
 
 

routers/auth.py

@@ -205,7 +205,7 @@ async def google_auth(
     access_token = create_access_token(user.user_id, user.email, user.token_version)
     
     # Sync DB to Drive (Async)
-    background_tasks.add_task(drive_service.upload_db)
+    background_tasks.add_task(drive_service.upload_db_async)
     
     return AuthResponse(
         success=True,
@@ -341,6 +341,6 @@ async def logout(
     await db.commit()
     
     # Sync DB to Drive (Async)
-    background_tasks.add_task(drive_service.upload_db)
+    background_tasks.add_task(drive_service.upload_db_async)
     
     return {"success": True, "message": "Logged out successfully. All sessions invalidated."}

routers/payments.py

@@ -407,7 +407,7 @@ async def verify_payment(
         )
         
         # Sync DB to Drive (Async)
-        background_tasks.add_task(drive_service.upload_db)
+        background_tasks.add_task(drive_service.upload_db_async)
         
         return VerifyPaymentResponse(
             success=True,
@@ -502,7 +502,7 @@ async def razorpay_webhook(
                         )
                         
                         # Sync DB to Drive (Async)
-                        background_tasks.add_task(drive_service.upload_db)
+                        background_tasks.add_task(drive_service.upload_db_async)
         
         # Handle payment failed event
         elif event == "payment.failed":

services/drive_service.py

@@ -1,6 +1,7 @@
 import os
 import logging
 import io
+import asyncio
 from dotenv import load_dotenv
 
 load_dotenv()
@@ -132,6 +133,10 @@ class DriveService:
             logger.error(f"Unexpected error uploading DB: {e}")
             return False
 
+    async def upload_db_async(self):
+        """Async wrapper for upload_db to run in a separate thread."""
+        return await asyncio.to_thread(self.upload_db)
+
     def download_db(self):
         """Download the database file from Google Drive."""
         if not self.service and not self.authenticate():

services/gemini_job_worker.py

@@ -45,14 +45,6 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
     def __init__(self):
         self.drive_service = DriveService()
         
-    async def _sync_db_to_drive(self):
-        """Sync database to Google Drive in a separate thread."""
-        try:
-            # Run synchronous upload_db in a thread to avoid blocking the event loop
-            await asyncio.to_thread(self.drive_service.upload_db)
-        except Exception as e:
-            logger.error(f"Failed to sync DB to Drive after job completion: {e}")
-
     async def _get_service_with_key(self, session: AsyncSession) -> tuple:
         """Get a GeminiService with the least-used API key."""
         from services.api_key_manager import get_least_used_key
@@ -196,7 +188,7 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
                         job.completed_at = datetime.utcnow()
                         success = True
                         # Sync DB on success
-                        await self._sync_db_to_drive()
+                        await self.drive_service.upload_db_async()
                     else:
                         job.status = "failed"
                         job.error_message = "No video URL returned"
@@ -299,7 +291,7 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.api_response = {"status": "success", "type": "image_edit"}
         job.completed_at = datetime.utcnow()
         # Sync DB on success
-        await self._sync_db_to_drive()
+        await self.drive_service.upload_db_async()
         return job
     
     async def _process_text(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
@@ -313,7 +305,7 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.api_response = {"result": result}
         job.completed_at = datetime.utcnow()
         # Sync DB on success
-        await self._sync_db_to_drive()
+        await self.drive_service.upload_db_async()
         return job
     
     async def _process_analyze(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
@@ -328,7 +320,7 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.api_response = {"result": result}
         job.completed_at = datetime.utcnow()
         # Sync DB on success
-        await self._sync_db_to_drive()
+        await self.drive_service.upload_db_async()
         return job
     
     async def _process_animation_prompt(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
@@ -343,7 +335,7 @@ class GeminiJobProcessor(JobProcessor[GeminiJob]):
         job.api_response = {"result": result}
         job.completed_at = datetime.utcnow()
         # Sync DB on success
-        await self._sync_db_to_drive()
+        await self.drive_service.upload_db_async()
         return job