Fix linting

Author: cyab05

python/rubin/rag/chatbot.py

@@ -170,12 +170,13 @@ def handle_user_input(
                 },
                 {"callbacks": [stream_handler]},
             )
-            msgs.add_ai_message(result["answer"]) # type: ignore[index]
+            msgs.add_ai_message(result["answer"])  # type: ignore[index]
 
             # Display source documents in an expander
             with st.expander("See sources"):
                 scores = [
-                    chunk.metadata["score"] for chunk in result["context"] # type: ignore[index]
+                    chunk.metadata["score"]
+                    for chunk in result["context"]  # type: ignore[index]
                 ]
 
                 if scores:
@@ -184,10 +185,10 @@ def handle_user_input(
                         max_score * 0.9
                     )  # Set threshold to 90% of the highest score
 
-                    for chunk in result["context"]: # type: ignore[index]
+                    for chunk in result["context"]:  # type: ignore[index]
                         score = chunk.metadata["score"]
 
                         # Only show sources with scores
                         # significantly higher (above the threshold)
                         if score >= threshold:
-                            st.info(f"Source: {chunk.metadata['source']}")                            
+                            st.info(f"Source: {chunk.metadata['source']}")

python/rubin/rag/extra_scripts/.DS_Store

@@ -2,12 +2,11 @@
 
 import json
 import time
-from pathlib import Path
 from functools import reduce
-from typing import Any, Dict, Optional
+from pathlib import Path
+from typing import Any
 
 import requests
-from typing import Optional
 
 
 def get_jira_issue(issue_name: str, email: str, api_token: str) -> tuple:
@@ -108,11 +107,14 @@ def extract_parent_issue(jira_data: dict) -> dict:
     return {}
 
 
-def safe_get(dictionary: Dict[str, Any], keys: list, default: Optional[Any] = None) -> Any:
+def safe_get(
+    dictionary: dict[str, Any], keys: list, default: Any | None = None
+) -> Any:
+    """Safely extract information from nested dictionaries."""
     return reduce(
-        lambda d, key: d.get(key, default) if isinstance(d, dict) else default, 
-        keys, 
-        dictionary
+        lambda d, key: d.get(key, default) if isinstance(d, dict) else default,
+        keys,
+        dictionary,
     )
 
 

python/rubin/rag/extra_scripts/parse_jira_tickets.py

@@ -24,7 +24,6 @@ def push_docs_to_weaviate(
     openai_api_key = os.getenv("OPENAI_API_KEY")
     if openai_api_key is None:
         raise ValueError("OPENAI_API_KEY environment variable is not set")
-    
 
     client = weaviate.connect_to_custom(
         http_host="localhost",
@@ -33,9 +32,7 @@ def push_docs_to_weaviate(
         grpc_host="localhost",
         grpc_port=50051,
         grpc_secure=False,
-        headers={
-            "X-OpenAI-Api-Key": openai_api_key  # Or any other inference API keys
-        },
+        headers={"X-OpenAI-Api-Key": openai_api_key},
     )
 
     if do_chunk: