Fix mypy errors and restore full type check

biswasbiplob · biswasbiplob · commit d7ef2833e56d · 2026-02-03T15:39:48.000+01:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -26,7 +26,7 @@ jobs:
         run: ruff check .
 
       - name: Type check
-        run: mypy feedback_agent/json_utils.py
+        run: mypy
 
       - name: Tests
         run: pytest -q
diff --git a/feedback_agent/agent.py b/feedback_agent/agent.py
@@ -240,14 +240,15 @@ def __init__(
 
         # Initialize plugins
         self.plugins = []
+        self.logging_plugin: Optional[LoggingPlugin] = None
+        self.metrics_plugin: Optional[ExamMetricsPlugin] = None
 
         # Add LoggingPlugin for ADK built-in observability (optional)
         if enable_logging_plugin:
             self.logging_plugin = LoggingPlugin()
             self.plugins.append(self.logging_plugin)
             logger.info("📝 LoggingPlugin enabled for ADK observability")
         else:
-            self.logging_plugin = None
             logger.info("📝 LoggingPlugin disabled (evaluation mode)")
 
         # Add custom metrics plugin if enabled
@@ -257,8 +258,6 @@ def __init__(
             )
             self.plugins.append(self.metrics_plugin)
             logger.info(f"📊 ExamMetricsPlugin enabled (logging to {metrics_file})")
-        else:
-            self.metrics_plugin = None
 
         # Create App with context compaction and plugins
         self.app = App(
diff --git a/feedback_agent/agents/analysis_agent.py b/feedback_agent/agents/analysis_agent.py
@@ -1,5 +1,5 @@
 from google.adk.agents.llm_agent import Agent
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
 from google.genai import types
 import json
 
@@ -16,7 +16,7 @@
 from feedback_agent.custom_llm import CustomGemini
 
 class AnalysisAgent:
-    def __init__(self, model: str = None):
+    def __init__(self, model: Optional[str] = None):
         # Get model from environment or use provided value
         if model is None:
             model = os.getenv('MODEL_NAME')
@@ -127,7 +127,7 @@ def __init__(self, model: str = None):
     def analyze_performance(
         self,
         graded_exam: Dict[str, Any],
-        exam_content: str = None
+        exam_content: Optional[str] = None
     ) -> Dict[str, Any]:
         """
         Analyze a graded exam to identify student weaknesses.
diff --git a/feedback_agent/agents/grading_agent.py b/feedback_agent/agents/grading_agent.py
@@ -1,6 +1,6 @@
 
 from google.adk.agents.llm_agent import Agent
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 from google.genai import types
 
 import logging
@@ -17,7 +17,7 @@
 from feedback_agent.json_utils import parse_json_payload
 
 class GradingAgent:
-    def __init__(self, model: str = None):
+    def __init__(self, model: Optional[str] = None):
         # Get model from environment or use provided value
         if model is None:
             model = os.getenv('MODEL_NAME')
diff --git a/feedback_agent/agents/image_processing_agent.py b/feedback_agent/agents/image_processing_agent.py
@@ -61,7 +61,7 @@ class ImageProcessingAgent:
 If no answer key is visible, set "answer_key" to "Not found".
 """
 
-    def __init__(self, model: str = None):
+    def __init__(self, model: Optional[str] = None):
         """
         Initialize the image processing agent.
 
@@ -192,6 +192,12 @@ async def process_image(
                     logger.info("Successfully extracted exam content from image")
                     return result
                 logger.warning("Failed to parse JSON from image processing response")
+                return {
+                    "subject": "Unknown",
+                    "exam_content": response_text,
+                    "answer_key": "Not found",
+                    "error": "Failed to parse JSON response",
+                }
             else:
                 logger.warning("Empty response from image processing")
                 return {
diff --git a/feedback_agent/agents/recommendation_agent.py b/feedback_agent/agents/recommendation_agent.py
@@ -1,5 +1,5 @@
 from google.adk.agents.llm_agent import Agent
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
 from google.genai import types
 import json
 
@@ -17,7 +17,7 @@
 from feedback_agent.json_utils import parse_json_payload
 
 class RecommendationAgent:
-    def __init__(self, model: str = None):
+    def __init__(self, model: Optional[str] = None):
         # Get model from environment or use provided value
         if model is None:
             model = os.getenv('MODEL_NAME')
diff --git a/feedback_agent/authorization.py b/feedback_agent/authorization.py
@@ -115,9 +115,10 @@ def get_my_performance(tool_context: ToolContext) -> Dict[str, Any]:
     all_weaknesses = []
     for exam in exams:
         exam_id = exam.get("exam_id")
-        analysis = _db.get_analysis(exam_id)
-        if analysis and analysis.get("weaknesses"):
-            all_weaknesses.extend(analysis["weaknesses"])
+        if exam_id:
+            analysis = _db.get_analysis(exam_id)
+            if analysis and analysis.get("weaknesses"):
+                all_weaknesses.extend(analysis["weaknesses"])
 
     return {
         "status": "success",
@@ -215,12 +216,13 @@ def get_student_performance(
     all_recommendations = []
     for exam in exams:
         exam_id = exam.get("exam_id")
-        analysis = _db.get_analysis(exam_id)
-        if analysis:
-            if analysis.get("weaknesses"):
-                all_weaknesses.extend(analysis["weaknesses"])
-            if analysis.get("recommendations"):
-                all_recommendations.append(analysis["recommendations"])
+        if exam_id:
+            analysis = _db.get_analysis(exam_id)
+            if analysis:
+                if analysis.get("weaknesses"):
+                    all_weaknesses.extend(analysis["weaknesses"])
+                if analysis.get("recommendations"):
+                    all_recommendations.append(analysis["recommendations"])
 
     return {
         "status": "success",
@@ -312,9 +314,10 @@ def get_class_statistics(tool_context: ToolContext) -> Dict[str, Any]:
             # Collect weaknesses
             for exam in exams:
                 exam_id = exam.get("exam_id")
-                analysis = _db.get_analysis(exam_id)
-                if analysis and analysis.get("weaknesses"):
-                    all_class_weaknesses.extend(analysis["weaknesses"])
+                if exam_id:
+                    analysis = _db.get_analysis(exam_id)
+                    if analysis and analysis.get("weaknesses"):
+                        all_class_weaknesses.extend(analysis["weaknesses"])
 
     # Calculate class averages
     if student_performances:
@@ -327,7 +330,7 @@ def get_class_statistics(tool_context: ToolContext) -> Dict[str, Any]:
         lowest_performer = None
 
     # Find most common weaknesses
-    weakness_counts = {}
+    weakness_counts: Dict[str, int] = {}
     for weakness in all_class_weaknesses:
         topic = weakness.get("topic", "Unknown")
         weakness_counts[topic] = weakness_counts.get(topic, 0) + 1
diff --git a/feedback_agent/conversational_tools.py b/feedback_agent/conversational_tools.py
@@ -718,16 +718,16 @@ def get_student_results(
                 "message": f"Student not found: {student_name or student_id}",
             }
 
-        student_id = student["student_id"]
-        student_name = student["name"]
-        exams = db.get_student_exams(student_id)
+        student_id_value: str = student["student_id"]
+        student_name_value: str = student["name"]
+        exams = db.get_student_exams(student_id_value)
 
         if not exams:
             return {
                 "status": "success",
-                "message": f"No exams recorded for {student_name}.",
-                "student_name": student_name,
-                "student_id": student_id,
+                "message": f"No exams recorded for {student_name_value}.",
+                "student_name": student_name_value,
+                "student_id": student_id_value,
                 "exams": [],
             }
 
@@ -749,8 +749,8 @@ def get_student_results(
 
         return {
             "status": "success",
-            "student_name": student_name,
-            "student_id": student_id,
+            "student_name": student_name_value,
+            "student_id": student_id_value,
             "summary": {
                 "total_exams": len(exams),
                 "average_percentage": round(avg_percentage, 1),
@@ -850,7 +850,7 @@ def get_class_analytics(tool_context: ToolContext) -> Dict[str, Any]:
         lowest = min(student_performances, key=lambda s: s["average_percentage"])
 
         # Find common weaknesses
-        weakness_counts = {}
+        weakness_counts: Dict[str, int] = {}
         for w in all_class_weaknesses:
             topic = w.get("topic", "Unknown")
             weakness_counts[topic] = weakness_counts.get(topic, 0) + 1
@@ -979,17 +979,24 @@ def get_learning_recommendations(
         db = system.db
 
         # Find student
-        if not student_id:
+        student_id_value: Optional[str] = student_id
+        if not student_id_value:
             student = db.get_student_by_name(student_name) if student_name else None
             if not student:
                 return {
                     "status": "error",
                     "message": f"Student not found: {student_name}",
                 }
-            student_id = student["student_id"]
+            student_id_value = student["student_id"]
+
+        if not student_id_value:
+            return {
+                "status": "error",
+                "message": "Student ID not available for recommendations.",
+            }
 
         # Get latest exam with recommendations
-        exams = db.get_student_exams(student_id)
+        exams = db.get_student_exams(student_id_value)
 
         if not exams:
             return {
diff --git a/feedback_agent/memory.py b/feedback_agent/memory.py
@@ -129,11 +129,13 @@ def get_recurring_weaknesses(
                 first_occurrence = min(occurrences, key=lambda o: o["index"])
                 last_occurrence = max(occurrences, key=lambda o: o["index"])
 
-                def _label(occ: Dict[str, Optional[str]]) -> str:
-                    if occ.get("date"):
-                        return occ["date"]
-                    if occ.get("exam_id"):
-                        return occ["exam_id"]
+                def _label(occ: Dict[str, Any]) -> str:
+                    date_value = occ.get("date")
+                    if date_value:
+                        return str(date_value)
+                    exam_id_value = occ.get("exam_id")
+                    if exam_id_value:
+                        return str(exam_id_value)
                     return f"exam_{occ['index']}"
 
                 pattern = WeaknessPattern(
diff --git a/feedback_agent/plugins.py b/feedback_agent/plugins.py
@@ -52,7 +52,7 @@ def __init__(self, log_to_file: bool = True, metrics_file: str = "exam_metrics.j
         # Aggregated statistics
         self.total_exams_processed = 0
         self.total_processing_time = 0.0
-        self.score_distribution = []
+        self.score_distribution: list[float] = []
         self.success_count = 0
         self.failure_count = 0
 
diff --git a/feedback_agent/utils.py b/feedback_agent/utils.py
@@ -31,12 +31,18 @@ async def delete_session(self, *, app_name: str, user_id: str, session_id: str)
 logger = logging.getLogger(__name__)
 
 
-def run_agent(agent: Agent, input_text: str, image_path: str = None, image_bytes: bytes = None, session_state: Dict[str, Any] = None) -> str:
+def run_agent(
+    agent: Agent,
+    input_text: str,
+    image_path: Optional[str] = None,
+    image_bytes: Optional[bytes] = None,
+    session_state: Optional[Dict[str, Any]] = None,
+) -> str:
     """
     Synchronous wrapper for running an agent.
     Handles creating a fresh event loop in a separate thread if needed.
     """
-    result_queue = queue.Queue()
+    result_queue: queue.Queue[object] = queue.Queue()
 
     def target():
         try:
@@ -56,9 +62,17 @@ def target():
     result = result_queue.get()
     if isinstance(result, Exception):
         raise result
+    if not isinstance(result, str):
+        raise TypeError(f"Expected string result, got {type(result)}")
     return result
 
-async def _run_agent_async(agent: Agent, prompt: str, image_path: str = None, image_bytes: bytes = None, session_state: Dict[str, Any] = None) -> str:
+async def _run_agent_async(
+    agent: Agent,
+    prompt: str,
+    image_path: Optional[str] = None,
+    image_bytes: Optional[bytes] = None,
+    session_state: Optional[Dict[str, Any]] = None,
+) -> str:
     # 1. Create Session
     session_id = str(uuid.uuid4())
     session = Session(id=session_id, app_name="feedback_agent", user_id="user")
diff --git a/pyproject.toml b/pyproject.toml
@@ -29,4 +29,4 @@ select = ["E9", "F63", "F7", "F82"]
 [tool.mypy]
 python_version = "3.13"
 ignore_missing_imports = true
-files = ["feedback_agent/json_utils.py"]
+files = ["feedback_agent"]
