Merge pull request #1671 from bluewave-labs/develop

Merging develop into master: June 18th 2025

Author: gorkem-bwl

.env.dev

@@ -6,6 +6,7 @@ FRONTEND_PORT=8082
 FRONTEND_URL=http://localhost:8082
 ALLOWED_ORIGINS=["http://localhost:5173", "http://localhost:8082"]
 BIAS_AND_FAIRNESS_PORT=8000
+NODE_ENV=development
 
 # Database
 DB_PORT=5432

.env.prod

@@ -6,6 +6,9 @@ FRONTEND_PORT=8080
 FRONTEND_URL=http://localhost:8080 # localhost needs to be replaced by ip
 ALLOWED_ORIGINS=["http://localhost:5173", "http://localhost:8080"] # localhost needs to be replaced by ip
 BIAS_AND_FAIRNESS_PORT=8000
+NODE_ENV=production
+# NOTE: Please do not change this unless the database is hosted on a different server and you want to use SSL for the connection
+DB_SSL=false
 
 # Database
 DB_PORT=5432

BiasAndFairnessServers/src/app.py

@@ -5,6 +5,7 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from routers.bias_and_fairness import router as bias_and_fairness
+from database.redis import close_redis
 from alembic.config import Config
 from alembic import command
 
@@ -19,7 +20,10 @@ def run_migrations():
     except Exception as e:
         logger.info(f"Error running migrations: {e}")
 
-app = FastAPI(on_startup=[run_migrations])
+async def shutdown_redis():
+    await close_redis()
+
+app = FastAPI(on_startup=[run_migrations], on_shutdown=[shutdown_redis])
 
 # enable CORS
 origins = [os.environ.get("BACKEND_URL") or "http://localhost:3000"]

BiasAndFairnessServers/src/controllers/bias_and_fairness.py

@@ -1,10 +1,14 @@
+import io
+import asyncio
 import json
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, Response
 from fastapi import HTTPException
 from crud.bias_and_fairness import upload_model, upload_data, insert_metrics, get_metrics_by_id, get_all_metrics_query, delete_metrics_by_id
 from utils.run_bias_and_fairness_check import analyze_fairness
+from utils.handle_files_uploads import process_files
 from database.db import get_db
-from fastapi import UploadFile
+from fastapi import UploadFile, BackgroundTasks
+from database.redis import get_next_job_id, get_job_status, delete_job_status
 
 async def get_all_metrics():
     """
@@ -60,67 +64,39 @@ async def get_metrics(id: int):
             detail=f"Failed to retrieve metrics, {str(e)}"
         )
 
-async def handle_upload(model: UploadFile, data: UploadFile, target_column: str, sensitive_column: str):
+async def get_upload_status(job_id: int):
+    value = await get_job_status(job_id)
+    if value is None:
+        return Response(status_code=204)
+    await delete_job_status(job_id)
+    return JSONResponse(
+        status_code=200,
+        content=value,
+        media_type="application/json"
+    )
+
+async def handle_upload(background_tasks: BackgroundTasks, model: UploadFile, data: UploadFile, target_column: str, sensitive_column: str):
     """
     Handle file upload from the client.
     """
-    try:
-        async with get_db() as db:
-            transaction = await db.begin()
-
-            model_content = await model.read()
-            model_filename = model.filename
-            data_content = await data.read()
-            data_filename = data.filename
-
-            if not model_content or not data_content:
-                raise ValueError("model or data file is empty")
-            if not model_filename or not data_filename:
-                raise ValueError("model or data file name is empty")
-
-            upload_model_record = await upload_model(content=model_content, name=model_filename, db=db)
-
-            if not upload_model_record:
-                raise Exception("failed to upload model file")
-
-            upload_data_record = await upload_data(
-                content=data_content,
-                name=data_filename,
-                target_column=target_column,
-                sensitive_column=sensitive_column,
-                model_id=upload_model_record.id,
-                db=db
-            )
-
-            if not upload_data_record:
-                raise Exception("failed to upload data file")
-            result = analyze_fairness(
-                model_content=model_content, 
-                data_content=data_content, 
-                target_column=target_column, 
-                sensitive_column=sensitive_column
-            )
-            
-            metrics = await insert_metrics(json.dumps(result), upload_data_record.id, db)
-            if not metrics:
-                raise Exception("failed to insert metrics")
-            
-            await transaction.commit()
-
-            return JSONResponse(
-                status_code=200,
-                content={
-                    "model_id": upload_model_record.id,
-                    "data_id": upload_data_record.id,
-                    "metrics_id": metrics.id,
-                    "metrics": result
-                }
-            )
-    except Exception as e:
-        raise HTTPException(
-            status_code=500,
-            detail=f"Failed to handle upload, {str(e)}"
-        )
+    job_id = await get_next_job_id()
+    response = JSONResponse(status_code=202, content={
+        "message": "Processing started", 
+        "job_id": job_id,
+        "model_filename": model.filename.replace(".gz", "") if model.filename else "",
+        "data_filename": data.filename.replace(".gz", "") if data.filename else ""
+    }, media_type="application/json")
+    model_ = {
+        "filename": model.filename,
+        "content": await model.read()
+    }
+    data_ = {
+        "filename": data.filename,
+        "content": await data.read()
+    }
+    # create a job ID or use a unique identifier for the task
+    background_tasks.add_task(process_files, job_id, model_, data_, target_column, sensitive_column)
+    return response
 
 async def delete_metrics(id: int):
     """
@@ -135,10 +111,7 @@ async def delete_metrics(id: int):
                     detail=f"Metrics with ID {id} not found"
                 )
             await db.commit()
-            return JSONResponse(
-                status_code=204,
-                content=None
-            )
+            return Response(status_code=204)
     except HTTPException as he:
         raise he
     except Exception as e:

BiasAndFairnessServers/src/database/redis.py

@@ -0,0 +1,43 @@
+import os
+import aioredis
+import json
+
+REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
+
+redis = None
+
+async def get_redis():
+    global redis
+    if not redis:
+        redis = await aioredis.from_url(REDIS_URL, decode_responses=True)
+    return redis
+
+async def close_redis():
+    global redis
+    if redis:
+        await redis.close()
+        redis = None
+
+async def get_next_job_id():
+    r = await get_redis()
+    # Use Redis INCR to generate unique job IDs
+    job_id = await r.incr("job_id_counter")
+    return job_id
+
+async def set_job_status(job_id: int, status: dict):
+    r = await get_redis()
+    # Store job status JSON serialized
+    await r.set(f"job_status:{job_id}", json.dumps(status), ex=3600)
+
+async def get_job_status(job_id: int):
+    r = await get_redis()
+    data = await r.get(f"job_status:{job_id}")
+    if data:
+        return json.loads(data)
+    return None
+
+async def delete_job_status(job_id: int):
+    r = await get_redis()
+    # Delete job status from Redis
+    await r.delete(f"job_status:{job_id}")
+    return True

BiasAndFairnessServers/src/requirements.txt

@@ -1,11 +1,11 @@
+aioredis==2.0.1
 alembic==1.16.1
 annotated-types==0.7.0
 anyio==4.9.0
 async-timeout==5.0.1
 asyncpg==0.30.0
 cffi==1.17.1
 click==8.1.8
-cryptography==45.0.3
 exceptiongroup==1.3.0
 fairlearn==0.12.0
 fastapi==0.115.12

BiasAndFairnessServers/src/routers/bias_and_fairness.py

@@ -1,22 +1,28 @@
-from fastapi import APIRouter, Form, UploadFile
-from controllers.bias_and_fairness import handle_upload as handle_upload_controller, get_metrics as get_metrics_controller, get_all_metrics as get_all_metrics_controller, delete_metrics as delete_metrics_controller
+from fastapi import APIRouter, Form, UploadFile, BackgroundTasks
+from controllers.bias_and_fairness import handle_upload as handle_upload_controller, get_metrics as get_metrics_controller, get_all_metrics as get_all_metrics_controller, delete_metrics as delete_metrics_controller, get_upload_status as get_upload_status_controller
 
 router = APIRouter()
 
 @router.post("/upload")
 async def upload_model(
+    background_tasks: BackgroundTasks,
     model: UploadFile = Form(...),
     data: UploadFile = Form(...),
     target_column: str = Form(...), 
     sensitive_column: str = Form(...), 
 ):
     return await handle_upload_controller(
+        background_tasks=background_tasks,
         model=model,
         data=data,
         target_column=target_column,
         sensitive_column=sensitive_column
     )
 
+@router.get("/upload/status/{job_id}")
+async def get_upload_status(job_id: int):
+    return await get_upload_status_controller(job_id)
+
 @router.get("/metrics/all")
 async def get_all_metrics():
     return await get_all_metrics_controller()

BiasAndFairnessServers/src/utils/handle_files_uploads.py

@@ -0,0 +1,83 @@
+import json
+import gzip
+import typing
+from fastapi import FastAPI, UploadFile
+from crud.bias_and_fairness import upload_model, upload_data, insert_metrics
+from utils.run_bias_and_fairness_check import analyze_fairness
+from database.db import get_db
+from database.redis import set_job_status
+
+@typing.no_type_check
+async def process_files(
+    job_id: int,
+    model: dict[str, typing.Union[bytes, str]],
+    data: dict[str, typing.Union[bytes, str]],
+    target_column: str,
+    sensitive_column: str,
+):
+    try:
+        async with get_db() as db:
+            transaction = await db.begin()
+
+            if not model["filename"] or not data["filename"]:
+                raise ValueError("model or data file name is empty")
+            model_filename = model["filename"].replace(".gz", "")
+            data_filename = data["filename"].replace(".gz", "")
+
+            if not model["content"] or not data["content"]:
+                raise ValueError("model or data file is empty")
+
+            model_bytes = model["content"]
+            data_bytes  = data["content"]
+
+            model_content = (
+                gzip.decompress(model_bytes) if model["filename"].endswith(".gz") else model_bytes
+            )
+            data_content = (
+                gzip.decompress(data_bytes) if data["filename"].endswith(".gz") else data_bytes
+            )
+
+            upload_model_record = await upload_model(content=model_content, name=model_filename, db=db)
+
+            if not upload_model_record:
+                raise Exception("failed to upload model file")
+
+            upload_data_record = await upload_data(
+                content=data_content,
+                name=data_filename,
+                target_column=target_column,
+                sensitive_column=sensitive_column,
+                model_id=upload_model_record.id,
+                db=db
+            )
+
+            if not upload_data_record:
+                raise Exception("failed to upload data file")
+            result = analyze_fairness(
+                model_content=model_content,
+                data_content=data_content,
+                target_column=target_column,
+                sensitive_column=sensitive_column
+            )
+
+            metrics = await insert_metrics(json.dumps(result), upload_data_record.id, db)
+            if not metrics:
+                raise Exception("failed to insert metrics")
+
+            await transaction.commit()
+            status = {
+                "status": "Completed",
+                "model_filename": model_filename,
+                "data_filename": data_filename,
+                "metrics_id": metrics.id,
+                "metrics": result
+            }
+    except Exception as e:
+        status = {
+            "status": "Failed", 
+            "error": str(e), 
+            "model_filename": locals().get("model_filename", "***"),
+            "data_filename": locals().get("data_filename", "***")
+        }
+
+    await set_job_status(job_id, status)

BiasAndFairnessServers/src/utils/unzip_files_async.py

@@ -0,0 +1,10 @@
+import functools
+import gzip
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+
+executor = ThreadPoolExecutor()
+
+async def decompress_async(data: bytes) -> bytes:
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(executor, gzip.decompress, data)

Clients/env.vars.ts

@@ -1,4 +1,8 @@
 export const ENV_VARs = {
-  URL: `${window.location.protocol}//${window.location.hostname}:3000` || 'http://localhost:3000/',
+  URL:
+    import.meta.env.VITE_APP_API_BASE_URL ??            // keep empty string if set
+    (typeof window !== 'undefined'                      // only run in browser
+      ? `${window.location.protocol}//${window.location.hostname}${window.location.protocol === 'https:' ? '' : ':3000'}` // use current URL if not set
+      : 'http://localhost:3000/'),                      // final Node/SSR fallback
   IS_DEMO_APP: import.meta.env.VITE_IS_DEMO_APP === 'true',
 };