fix3

app.py

@@ -11,7 +11,7 @@ from fastapi.responses import HTMLResponse
 from uvicorn import run as uvicorn_run
 
 # Slack Libraries (Async versions)
-from slack_bolt import AsyncApp
+from slack_bolt.async_app import AsyncApp
 from slack_bolt.adapter.fastapi import AsyncSlackRequestHandler
 from slack_sdk.oauth.installation_store.async_installation_store import AsyncInstallationStore
 from slack_sdk.oauth import AuthorizeUrlGenerator
@@ -67,7 +67,6 @@ class SupabaseAsyncInstallationStore(AsyncInstallationStore):
         self.supabase = supabase_client
 
     async def save(self, installation: Installation) -> None:
-        loop = asyncio.get_event_loop()
         data = {
             "team_id": installation.team_id,
             "bot_token": installation.bot_token,
@@ -76,21 +75,19 @@ class SupabaseAsyncInstallationStore(AsyncInstallationStore):
             "app_id": installation.app_id,
         }
         try:
-            await loop.run_in_executor(
-                None,
-                lambda: self.supabase.table("installations").upsert(data, on_conflict="team_id").execute()
+            await asyncio.to_thread(
+                self.supabase.table("installations").upsert, data, on_conflict="team_id"
             )
+            await asyncio.to_thread(self.supabase.table("installations").upsert(data, on_conflict="team_id").execute)
             logger.info(f"Saved installation for team: {installation.team_id}")
         except Exception as e:
             logger.error(f"Failed to save installation for team {installation.team_id}: {e}")
             raise
 
     async def fetch_installation(self, team_id: str, *, enterprise_id: str | None = None, user_id: str | None = None) -> Installation | None:
-        loop = asyncio.get_event_loop()
         try:
-            result = await loop.run_in_executor(
-                None,
-                lambda: self.supabase.table("installations").select("*").eq("team_id", team_id).execute()
+            result = await asyncio.to_thread(
+                self.supabase.table("installations").select("*").eq("team_id", team_id).execute
             )
             if not result.data:
                 return None
@@ -111,11 +108,9 @@ class SupabaseAsyncInstallationStore(AsyncInstallationStore):
             raise
 
     async def delete(self, team_id: str, *, enterprise_id: str | None = None, user_id: str | None = None) -> None:
-        loop = asyncio.get_event_loop()
         try:
-            await loop.run_in_executor(
-                None,
-                lambda: self.supabase.table("installations").delete().eq("team_id", team_id).execute()
+            await asyncio.to_thread(
+                self.supabase.table("installations").delete().eq("team_id", team_id).execute
             )
             logger.info(f"Deleted installation for team: {team_id}")
         except Exception as e:
@@ -209,44 +204,38 @@ def embed_text(text: str) -> List[float]:
     embedding = embedding_model.encode(text, convert_to_tensor=False)
     return embedding.tolist()
 
-def store_embeddings(chunks: List[str]):
+async def store_embeddings(chunks: List[str]):
     """Stores text chunks and their embeddings in Supabase."""
-    loop = asyncio.get_event_loop()
     for chunk in chunks:
         embedding = embed_text(chunk)
         try:
-            loop.run_in_executor(
-                None,
-                lambda c=chunk, e=embedding: supabase.table("documents").insert({
-                    "content": c,
-                    "embedding": e
-                }).execute()
+            await asyncio.to_thread(
+                supabase.table("documents").insert({
+                    "content": chunk,
+                    "embedding": embedding
+                }).execute
             )
         except Exception as e:
             logger.error(f"Failed to insert chunk into Supabase: {str(e)}")
 
-def is_table_empty() -> bool:
+async def is_table_empty() -> bool:
     """Checks if the documents table has any records."""
     try:
-        loop = asyncio.get_event_loop()
-        result = loop.run_in_executor(
-            None,
-            lambda: supabase.table("documents").select("id", count="exact").limit(1).execute()
+        result = await asyncio.to_thread(
+            supabase.table("documents").select("id", count="exact").limit(1).execute
         )
         return result.count == 0
     except Exception as e:
         logger.error(f"Error checking table emptiness: {str(e)}")
         return True
 
-def search_documents(query: str, match_count: int = 5) -> List[Dict[str, Any]]:
+async def search_documents(query: str, match_count: int = 5) -> List[Dict[str, Any]]:
     """Searches Supabase for documents matching the query using vector similarity."""
     if embedding_model is None:
         raise RuntimeError("Embedding model not loaded.")
     query_embedding = embed_text(query)
     try:
-        loop = asyncio.get_event_loop()
-        result = loop.run_in_executor(
-            None,
+        result = await asyncio.to_thread(
             lambda: supabase.rpc("match_documents", {
                 "query_embedding": query_embedding,
                 "match_count": match_count
@@ -257,7 +246,7 @@ def search_documents(query: str, match_count: int = 5) -> List[Dict[str, Any]]:
         logger.error(f"Error searching documents for query '{query}': {str(e)}")
         return []
 
-def answer_question(question: str, context: str) -> str:
+async def answer_question(question: str, context: str) -> str:
     """Answers a question based on the provided context using the QA pipeline."""
     if qa_pipeline is None:
         raise RuntimeError("QA pipeline not loaded.")
@@ -267,7 +256,9 @@ def answer_question(question: str, context: str) -> str:
         MAX_CONTEXT_LEN = 4096 
         context_slice = context[:MAX_CONTEXT_LEN]
         
-        result = qa_pipeline(question=question, context=context_slice)
+        result = await asyncio.to_thread(
+            qa_pipeline, question=question, context=context_slice
+        )
         return result['answer']
     except Exception as e:
         logger.error(f"Error in QA pipeline: {str(e)}")
@@ -291,13 +282,13 @@ async def handle_file_shared(event, say, client):
             return
         
         token = client.token 
-        file_content = download_slack_file(file_url, token)
+        file_content = await asyncio.to_thread(download_slack_file, file_url, token)
         
         text = ""
         if "pdf" in file_type:
-            text = extract_text_from_pdf(file_content)
+            text = await asyncio.to_thread(extract_text_from_pdf, file_content)
         elif "wordprocessingml" in file_type or "msword" in file_type:
-            text = extract_text_from_docx(file_content)
+            text = await asyncio.to_thread(extract_text_from_docx, file_content)
         else:
             await say("Unsupported file type. Please upload PDF or DOCX files.")
             return
@@ -307,7 +298,7 @@ async def handle_file_shared(event, say, client):
             return
         
         chunks = chunk_text(text)
-        store_embeddings(chunks)
+        await store_embeddings(chunks)
         
         await say(f"✅ File processed successfully! Added **{len(chunks)}** chunks to the knowledge base.")
     except Exception as e:
@@ -337,11 +328,11 @@ async def handle_mention(event, say, client):
     
     try:
         # Check if table is empty
-        if await asyncio.to_thread(is_table_empty):
+        if await is_table_empty():
             await say("My knowledge base is empty. Please share some PDF or DOCX files first so I can learn!")
             return
         
-        results = search_documents(user_query, match_count=5)
+        results = await search_documents(user_query, match_count=5)
         
         if not results:
             await say("I couldn't find any relevant information in my knowledge base. Try uploading more documents.")
@@ -349,7 +340,7 @@ async def handle_mention(event, say, client):
         
         # Combine the content of the top documents into a single context string
         context = " ".join([doc["content"] for doc in results])
-        answer = answer_question(user_query, context)
+        answer = await answer_question(user_query, context)
         
         # Format the final response
         await say(f"💡 *Answer:* {answer}\n\n_I found this information by analyzing {len(results)} relevant document chunks._")
@@ -381,10 +372,8 @@ async def health():
     """Health check endpoint."""
     db_status = "error"
     try:
-        loop = asyncio.get_event_loop()
-        await loop.run_in_executor(
-            None,
-            lambda: supabase.table("installations").select("team_id").limit(1).execute()
+        await asyncio.to_thread(
+            supabase.table("installations").select("team_id").limit(1).execute
         )
         db_status = "ok"
     except Exception as e: