refactor: backend main & search for namespaces

Author: adithayyil

backend/main.py

@@ -1,6 +1,6 @@
 import os
 import logging
-from fastapi import UploadFile, HTTPException
+from fastapi import UploadFile, HTTPException, Form
 import modal
 
 # Constants
@@ -110,16 +110,16 @@ def startup(self):
         print(f"[Container] Started at {self.start_time.isoformat()}")
 
     @modal.method()
-    async def process_video(self, video_bytes: bytes, filename: str, job_id: str):
-        logger.info(f"[Job {job_id}] Processing started: {filename} ({len(video_bytes)} bytes)")
+    async def process_video(self, video_bytes: bytes, filename: str, job_id: str, namespace: str = ""):
+        logger.info(f"[Job {job_id}] Processing started: {filename} ({len(video_bytes)} bytes) | namespace='{namespace}'")
 
         try:
             # Upload original video to R2 bucket
             # TODO: do this in parallel with processing and provide url once done
             success, hashed_identifier = self.r2_connector.upload_video(
                 video_data=video_bytes,
                 filename=filename,
-                # user_id="user1" # Specify user ID once we have user management
+                namespace=namespace
             )
             if not success:
                 raise Exception(f"Failed to upload video to R2 storage: {hashed_identifier}")
@@ -175,7 +175,7 @@ async def process_video(self, video_bytes: bytes, filename: str, job_id: str):
                 self.pinecone_connector.upsert_chunk(
                     chunk_id=chunk['chunk_id'],
                     chunk_embedding=embedding.numpy(),
-                    namespace="",
+                    namespace=namespace,
                     metadata=chunk['metadata']
                 )            
 
@@ -244,12 +244,13 @@ async def status(self, job_id: str):
         return job_data
 
     @modal.fastapi_endpoint(method="POST")
-    async def upload(self, file: UploadFile = None):
+    async def upload(self, file: UploadFile = None, namespace: str = Form("")):
         """
         Handle video file upload and start background processing.
 
         Args:
             file (UploadFile): The uploaded video file.
+            namespace (str, optional): Namespace for Pinecone and R2 storage (default: "")
 
         Returns:
             dict: Contains job_id, filename, content_type, size_bytes, status, and message.
@@ -264,11 +265,12 @@ async def upload(self, file: UploadFile = None):
             "filename": file.filename,
             "status": "processing",
             "size_bytes": file_size,
-            "content_type": file.content_type
+            "content_type": file.content_type,
+            "namespace": namespace
         })
 
         # Spawn background processing (non-blocking - returns immediately)
-        self.process_video.spawn(contents, file.filename, job_id)
+        self.process_video.spawn(contents, file.filename, job_id, namespace)
 
         return {
             "job_id": job_id,
@@ -281,12 +283,13 @@ async def upload(self, file: UploadFile = None):
 
 
     @modal.fastapi_endpoint(method="GET")
-    async def search(self, query: str):
+    async def search(self, query: str, namespace: str = ""):
         """
         Search endpoint - accepts a text query and returns semantic search results.
 
         Args:
         - query (str): The search query string (required)
+        - namespace (str, optional): Namespace for Pinecone search (default: "")
         - top_k (int, optional): Number of top results to return (default: 10)
 
         Returns: dict with 'query', 'results', and 'timing'.
@@ -300,16 +303,17 @@ async def search(self, query: str):
                 raise HTTPException(status_code=400, detail="Missing 'query' parameter")
 
             top_k = 10
-            logger.info(f"[Search] Query: '{query}' | top_k={top_k}")
+            logger.info(f"[Search] Query: '{query}' | namespace='{namespace}' | top_k={top_k}")
 
             # Execute semantic search
             results = self.searcher.search(
                 query=query,
-                top_k=top_k
+                top_k=top_k,
+                namespace=namespace
             )
-            
+
             t_done = time.perf_counter()
-            
+
             # Log chunk-level results only
             logger.info(f"[Search] Found {len(results)} chunk-level results in {t_done - t_start:.3f}s")
 

backend/search/searcher.py

@@ -61,14 +61,16 @@ def device(self) -> str:
     def search(
         self,
         query: str,
-        top_k: int = 5
+        top_k: int = 5,
+        namespace: Optional[str] = None
     ) -> List[Dict[str, Any]]:
         """
         Search for semantically similar content.
 
         Args:
             query: Natural language search query
             top_k: Number of results to return (default: 5)
+            namespace: Optional namespace to override default (default: None uses self.namespace)
 
         Returns:
             List of matches with scores and metadata, sorted by similarity
@@ -79,15 +81,17 @@ def search(
                 print(f"Score: {result['score']}")
                 print(f"Metadata: {result['metadata']}")
         """
-        logger.info(f"Searching for: '{query}' (top_k={top_k})")
-        
+        # Use provided namespace or fall back to default
+        search_namespace = namespace if namespace is not None else self.namespace
+        logger.info(f"Searching for: '{query}' (top_k={top_k}, namespace='{search_namespace}')")
+
         # Generate query embedding
         query_embedding = self.embedder.embed_text(query)
-        
+
         # Search Pinecone with optional filters
         matches = self.connector.query_chunks(
             query_embedding=query_embedding,
-            namespace=self.namespace,
+            namespace=search_namespace,
             top_k=top_k
         )