🚧 Working on upload and download endpoints post PR review

Signed-off-by: m-misiura <[email protected]>

Author: dependabot[bot]

.github/workflows/python-tests.yaml

@@ -2,7 +2,7 @@ name: Python Tests
 
 on:
   push:
-    branches: [main, download_and_upload_endpoints]
+    branches: [main]
   pull_request:
     branches: [main]
 

pyproject.toml

@@ -23,16 +23,11 @@ dev = [
     "isort>=5.12.0,<7",
     "flake8>=6.1.0,<8",
     "mypy>=1.5.1,<2",
-<<<<<<< HEAD
     "pytest-cov>=4.1.0,<5",
     "httpx>=0.25.0,<0.29",
-=======
-    "pytest-cov>=4.1.0,<7",
-    "httpx>=0.25.0,<0.26",
->>>>>>> 331eed9 (Update pytest-cov requirement from <5,>=4.1.0 to >=4.1.0,<7)
 ]
 eval = ["lm-eval[api]==0.4.4", "fastapi-utils>=0.8.0", "typing-inspect==0.9.0"]
-protobuf = ["numpy>=1.24.0,<2", "grpcio>=1.62.1,<2", "grpcio-tools>=1.62.1,<2"]
+protobuf = ["numpy>=1.24.0,<3", "grpcio>=1.62.1,<2", "grpcio-tools>=1.62.1,<2"]
 
 [tool.hatch.build.targets.sdist]
 include = ["src"]

src/endpoints/data/data_download.py

@@ -6,7 +6,7 @@
 from src.service.utils.download import (
     DataRequestPayload,
     DataResponsePayload,
-    apply_matcher,
+    apply_filters,  # ← New utility function
     load_model_dataframe,
 )
 
@@ -19,35 +19,10 @@ async def download_data(payload: DataRequestPayload) -> DataResponsePayload:
     """Download model data with filtering."""
     try:
         logger.info(f"Received data download request for model: {payload.modelId}")
-
-        # Load the dataframe
         df = await load_model_dataframe(payload.modelId)
-
         if df.empty:
             return DataResponsePayload(dataCSV="")
-        # Apply matchAll filters (AND logic)
-        if payload.matchAll:
-            for matcher in payload.matchAll:
-                df = apply_matcher(df, matcher, negate=False)
-        # Apply matchNone filters (NOT logic)
-        if payload.matchNone:
-            for matcher in payload.matchNone:
-                df = apply_matcher(df, matcher, negate=True)
-        base_df = df.copy()
-        # Apply matchAny filters (OR logic)
-        if payload.matchAny:
-            matching_dfs = []
-            for matcher in payload.matchAny:
-                matched_df = apply_matcher(base_df, matcher, negate=False)
-                if not matched_df.empty:
-                    matching_dfs.append(matched_df)
-            # Union all results
-            if matching_dfs:
-                df = pd.concat(matching_dfs, ignore_index=True).drop_duplicates()
-            else:
-                # No matches found, return empty dataframe with same columns
-                df = pd.DataFrame(columns=df.columns)
-        # Convert to CSV
+        df = apply_filters(df, payload)
         csv_data = df.to_csv(index=False)
         return DataResponsePayload(dataCSV=csv_data)
     except HTTPException:

src/endpoints/data/data_upload.py

@@ -10,104 +10,30 @@
 from src.service.constants import INPUT_SUFFIX, METADATA_SUFFIX, OUTPUT_SUFFIX
 from src.service.data.modelmesh_parser import ModelMeshPayloadParser
 from src.service.data.storage import get_storage_interface
-from src.service.utils.upload import (
-    handle_ground_truths,
-    process_tensors,
-    sanitize_id,
-    save_model_data,
-    validate_data_tag,
-    validate_input_shapes,
-)
+from src.service.utils.upload import process_upload_request
 
 router = APIRouter()
 logger = logging.getLogger(__name__)
 
+
 class UploadPayload(BaseModel):
     model_name: str
     data_tag: Optional[str] = None
     is_ground_truth: bool = False
     request: Dict[str, Any]
-    response: Dict[str, Any]
+    response:  Optional[Dict[str, Any]] = None
 
 
 @router.post("/data/upload")
 async def upload(payload: UploadPayload) -> Dict[str, str]:
     """Upload model data - regular or ground truth."""
     try:
-        # Get fresh storage interface for each request
-        storage = get_storage_interface()
-        
-        model_name = ModelMeshPayloadParser.standardize_model_id(payload.model_name)
-        if payload.data_tag and (error := validate_data_tag(payload.data_tag)):
-            raise HTTPException(400, error)
-        inputs = payload.request.get("inputs", [])
-        outputs = payload.response.get("outputs", [])
-        if not inputs or not outputs:
-            raise HTTPException(400, "Missing input or output tensors")
-        input_arrays, input_names, _, execution_ids = process_tensors(inputs)
-        output_arrays, output_names, _, _ = process_tensors(outputs)
-        if error := validate_input_shapes(input_arrays, input_names):
-            raise HTTPException(400, f"One or more errors in input tensors: {error}")
-        
-        if payload.is_ground_truth:
-            if not execution_ids:
-                raise HTTPException(400, "Ground truth requires execution IDs")
-            result = await handle_ground_truths(
-                model_name,
-                input_arrays,
-                input_names,
-                output_arrays,
-                output_names,
-                [sanitize_id(id) for id in execution_ids],
-            )
-            if not result.success:
-                raise HTTPException(400, result.message)
-            result_data = result.data
-            if result_data is None:
-                raise HTTPException(500, "Ground truth processing failed")
-            gt_name = f"{model_name}_ground_truth"
-            await storage.write_data(gt_name + OUTPUT_SUFFIX, result_data["outputs"], result_data["output_names"])
-            await storage.write_data(
-                gt_name + METADATA_SUFFIX,
-                result_data["metadata"],
-                result_data["metadata_names"],
-            )
-            return {"message": result.message}
-        else:
-            n_rows = input_arrays[0].shape[0]
-            exec_ids = execution_ids or [str(uuid.uuid4()) for _ in range(n_rows)]
-
-            def flatten(arrays: List[np.ndarray], row: int) -> List[Any]:
-                return [x for arr in arrays for x in (arr[row].flatten() if arr.ndim > 1 else [arr[row]])]
-
-            input_data = [flatten(input_arrays, i) for i in range(n_rows)]
-            output_data = [flatten(output_arrays, i) for i in range(n_rows)]
-            cols = ["id", "model_id", "timestamp", "tag"]
-            current_timestamp = datetime.now().isoformat()
-            metadata_rows = [
-                [
-                    str(eid),
-                    str(model_name),
-                    str(current_timestamp),
-                    str(payload.data_tag or ""),
-                ]
-                for eid in exec_ids
-            ]
-            metadata = np.array(metadata_rows, dtype="<U100")
-            await save_model_data(
-                model_name,
-                np.array(input_data),
-                input_names,
-                np.array(output_data),
-                output_names,
-                metadata,
-                cols,
-            )
-            return {"message": f"{n_rows} datapoints added to {model_name}"}
-            
+        logger.info(f"Received upload request for model: {payload.model_name}")
+        result = await process_upload_request(payload)
+        logger.info(f"Upload completed for model: {payload.model_name}")
+        return result
     except HTTPException:
-        # Re-raise HTTP exceptions as-is
         raise
     except Exception as e:
         logger.error(f"Unexpected error in upload endpoint for model {payload.model_name}: {str(e)}", exc_info=True)
-        raise HTTPException(500, f"Internal server error: {str(e)}")
+        raise HTTPException(500, f"Internal server error: {str(e)}")