Upload folder using huggingface_hub

database.py

@@ -84,6 +84,21 @@ def init_db():
             FOREIGN KEY(owner_username) REFERENCES users(username)
         )
     ''')
+
+    # Create Jobs Table (PERSISTENCE)
+    c.execute('''
+        CREATE TABLE IF NOT EXISTS jobs (
+            id TEXT PRIMARY KEY,
+            status TEXT NOT NULL,
+            result TEXT, -- JSON serialized
+            error TEXT,
+            created_at REAL,
+            storage_path TEXT,
+            username TEXT,
+            file_type TEXT,
+            FOREIGN KEY(username) REFERENCES users(username)
+        )
+    ''')
     
     conn.commit()
     conn.close()
@@ -349,6 +364,7 @@ def update_patient(username: str, patient_db_id: int, updates: Dict[str, Any]) -
         values.extend([patient_db_id, username])
         query = f"UPDATE patients SET {', '.join(fields)} WHERE id = ? AND owner_username = ?"
         
+
         c.execute(query, values)
         count = c.rowcount
         conn.commit()
@@ -358,4 +374,79 @@ def update_patient(username: str, patient_db_id: int, updates: Dict[str, Any]) -
         logging.error(f"Error updating patient: {e}")
         return False
 
+# --- Job Operations (Persistence) ---
+
+import json
+
+def create_job(job_data: Dict[str, Any]):
+    """Create a new job record."""
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        c.execute('''
+            INSERT INTO jobs (id, status, result, error, created_at, storage_path, username, file_type)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        ''', (
+            job_data['id'],
+            job_data.get('status', 'pending'),
+            json.dumps(job_data.get('result')) if job_data.get('result') else None,
+            job_data.get('error'),
+            job_data['created_at'],
+            job_data.get('storage_path'),
+            job_data.get('username'),
+            job_data.get('file_type')
+        ))
+        conn.commit()
+        conn.close()
+        return True
+    except Exception as e:
+        logging.error(f"Error creating job: {e}")
+        return False
+
+def get_job(job_id: str) -> Optional[Dict[str, Any]]:
+    """Retrieve job by ID."""
+    conn = get_db_connection()
+    c = conn.cursor()
+    c.execute('SELECT * FROM jobs WHERE id = ?', (job_id,))
+    row = c.fetchone()
+    conn.close()
+    
+    if row:
+        job = dict(row)
+        if job['result']:
+            try:
+                job['result'] = json.loads(job['result'])
+            except:
+                job['result'] = None
+        return job
+    return None
+
+def update_job_status(job_id: str, status: str, result: Optional[Dict] = None, error: Optional[str] = None):
+    """Update job status and result."""
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        
+        updates = ["status = ?"]
+        params = [status]
+        
+        if result is not None:
+            updates.append("result = ?")
+            params.append(json.dumps(result))
+            
+        if error is not None:
+            updates.append("error = ?")
+            params.append(error)
+            
+        params.append(job_id)
+        
+        query = f"UPDATE jobs SET {', '.join(updates)} WHERE id = ?"
+        c.execute(query, params)
+        conn.commit()
+        conn.close()
+        return True
+    except Exception as e:
+        logging.error(f"Error updating job: {e}")
+        return False
+
 

main.py

@@ -699,7 +699,7 @@ class FeedbackModel(BaseModel):
 # =========================================================================
 # GLOBAL STATE
 # =========================================================================
-jobs: Dict[str, Job] = {}
+jobs: Dict[str, Job] = {}  # REMOVED: Now using SQLite persistence
 storage_provider = get_storage_provider(os.getenv("STORAGE_MODE", "LOCAL"))
 
 # Initialize Database
@@ -1232,12 +1232,20 @@ async def limit_concurrency(request: Request, call_next):
 # =========================================================================
 async def process_analysis(job_id: str, image_bytes: bytes):
     """Background task to run inference and log to registry."""
-    job = jobs.get(job_id)
+    # RESILIENCE: Retrieve job from DB
+    job = database.get_job(job_id)
     if not job:
+        logger.error(f"❌ Job {job_id} not found in DB during background processing")
         return
+    
+    # We must construct a Job object or just work with the dict
+    # Let's work with the dict for consistency, or simple variables
+    username = job.get('username')
+    file_type = job.get('file_type')
 
     logger.info(f"Processing Job {job_id}")
-    job.status = JobStatus.PROCESSING
+    database.update_job_status(job_id, JobStatus.PROCESSING.value)
+    
     start_time = time.time()
     
     try:
@@ -1247,27 +1255,27 @@ async def process_analysis(job_id: str, image_bytes: bytes):
         loop = asyncio.get_event_loop()
         result = await loop.run_in_executor(None, model_wrapper.predict, image_bytes)
         
-        job.result = result
-        job.status = JobStatus.COMPLETED
-        
         # Calculate computation time
         computation_time_ms = int((time.time() - start_time) * 1000)
         
+        # Update Job in DB
+        database.update_job_status(job_id, JobStatus.COMPLETED.value, result=result)
+        
         # Log to registry (REAL DATA)
-        if job.username and result:
+        if username and result:
             domain = result.get('domain', {}).get('label', 'Unknown')
             top_diag = result.get('specific', [{}])[0].get('label', 'Unknown') if result.get('specific') else 'Unknown'
             confidence = result.get('specific', [{}])[0].get('probability', 0) if result.get('specific') else 0
             priority = result.get('priority', 'Normale')
             
             database.log_analysis(
-                username=job.username,
+                username=username,
                 domain=domain,
                 top_diagnosis=top_diag,
                 confidence=confidence,
                 priority=priority,
                 computation_time_ms=computation_time_ms,
-                file_type=job.file_type or 'Unknown'
+                file_type=file_type or 'Unknown'
             )
             logger.info(f"✅ Job {job_id} logged to registry")
         
@@ -1275,8 +1283,7 @@ async def process_analysis(job_id: str, image_bytes: bytes):
         
     except Exception as e:
         logger.error(f"❌ Job {job_id} failed: {str(e)}")
-        job.error = str(e)
-        job.status = JobStatus.FAILED
+        database.update_job_status(job_id, JobStatus.FAILED.value, error=str(e))
 
 # =========================================================================
 # API ENDPOINTS
@@ -1403,15 +1410,16 @@ async def analyze_image(
     else:
         file_type = 'OTHER'
 
-    jobs[job_id] = Job(
-        id=job_id, 
-        status=JobStatus.PENDING, 
-        created_at=time.time(),
-        encrypted_user=enc_user,
-        storage_path=storage_path,
-        username=current_user.username,  # For registry logging
-        file_type=file_type  # For registry logging
-    )
+    # Persist Job to DB
+    job_data = {
+        "id": job_id,
+        "status": JobStatus.PENDING.value,
+        "created_at": time.time(),
+        "storage_path": storage_path,
+        "username": current_user.username,
+        "file_type": file_type
+    }
+    database.create_job(job_data)
     
     background_tasks.add_task(process_analysis, job_id, image_bytes)
     
@@ -1425,19 +1433,19 @@ async def get_result(task_id: str, current_user: User = Depends(get_current_user
     - **Requires authentication**
     - Returns job status and results when complete
     """
-    job = jobs.get(task_id)
+    job = database.get_job(task_id)
     if not job:
         logger.warning(f"Job not found: {task_id}")
+        # If job is lost (server restart before persistence, or bad ID), return 404
+        # Frontend should handle this by stopping polling
         raise HTTPException(status_code=404, detail="Job not found")
     
-    # Verify ownership: decrypt stored user and compare
-    if job.encrypted_user:
-        stored_user = encryption.decrypt_data(job.encrypted_user)
-        if stored_user != current_user.username:
-            logger.warning(f"Unauthorized access attempt to job {task_id} by {current_user.username}")
-            raise HTTPException(status_code=403, detail="Access denied")
+    # Verify ownership
+    if job.get('username') != current_user.username:
+        logger.warning(f"Unauthorized access attempt to job {task_id} by {current_user.username}")
+        raise HTTPException(status_code=403, detail="Access denied")
     
-    logger.info(f"Polling Job {task_id}: Status={job.status}")
+    logger.info(f"Polling Job {task_id}: Status={job.get('status')}")
     return job
 
 @app.get("/health")

upload_space.py

@@ -17,5 +17,5 @@ upload_folder(
     ignore_patterns=["models/*", "*.pyc", "__pycache__", "*.db", "storage/*", "data_storage/*", ".env", "venv", ".git", ".idea"]
 )
 
-print("✅ Upload complete!")
+print("[OK] Upload complete!")
 print("Your Space should start building at: https://huggingface.co/spaces/issoufzousko07/elephmind-api")