refactor: Fix type annotations and null checks in LMEval classes (#52)

Author: ruivieira

src/llama_stack_provider_lmeval/lmeval.py

@@ -325,7 +325,7 @@ def __init__(self, namespace: str = "default", service_account: str | None = Non
         """
         self._namespace = namespace
         self._service_account = service_account
-        self._config = None
+        self._config: LMEvalEvalProviderConfig | None = None
 
     @staticmethod
     def _build_openai_url(base_url: str) -> str:
@@ -347,7 +347,7 @@ def _build_openai_url(base_url: str) -> str:
             return f"{cleaned_url}/v1/completions"
 
     def _create_model_args(
-        self, base_url: str, benchmark_config: BenchmarkConfig
+        self, base_url: str | None, benchmark_config: BenchmarkConfig
     ) -> list[ModelArg]:
         """Create model arguments for the LMEvalJob CR."""
         model_args = [
@@ -749,7 +749,8 @@ def create_cr(
         model_args = self._create_model_args(base_url, task_config)
 
         if (
-            hasattr(stored_benchmark, "metadata")
+            stored_benchmark is not None
+            and hasattr(stored_benchmark, "metadata")
             and stored_benchmark.metadata
             and "tokenizer" in stored_benchmark.metadata
         ):
@@ -762,7 +763,8 @@ def create_cr(
 
         # Add tokenized_requests parameter if present in metadata
         if (
-            hasattr(stored_benchmark, "metadata")
+            stored_benchmark is not None
+            and hasattr(stored_benchmark, "metadata")
             and stored_benchmark.metadata
             and "tokenized_requests" in stored_benchmark.metadata
         ):
@@ -832,7 +834,7 @@ def create_cr(
         if git_source_data:
             logger.info("Adding customTasks to CR with git data: %s", git_source_data)
 
-            custom_tasks_section = {"source": {"git": {}}}
+            custom_tasks_section: dict[str, Any] = {"source": {"git": {}}}
 
             for key, value in git_source_data.items():
                 if value is not None:
@@ -926,12 +928,12 @@ def __init__(self, config: LMEvalEvalProviderConfig):
 
         logger.debug("LMEval provider initialized with namespace: %s", self._namespace)
         logger.debug("LMEval provider config values: %s", vars(self._config))
-        self.benchmarks = {}
+        self.benchmarks: dict[str, Benchmark] = {}
         self._jobs: list[Job] = []
-        self._job_metadata = {}
+        self._job_metadata: dict[str, dict[str, Any]] = {}
 
-        self._k8s_client = None
-        self._k8s_custom_api = None
+        self._k8s_client: k8s_client.ApiClient | None = None
+        self._k8s_custom_api: k8s_client.CustomObjectsApi | None = None
         if self.use_k8s:
             self._init_k8s_client()
             logger.debug(
@@ -1045,8 +1047,10 @@ def _deploy_lmeval_cr(self, cr: dict, job_id: str) -> None:
         if "spec" in cr:
             pvc_name = None
 
-            if hasattr(self._cr_builder, "_config") and hasattr(
-                self._cr_builder._config, "metadata"
+            if (
+                self._cr_builder._config is not None
+                and hasattr(self._cr_builder._config, "metadata")
+                and self._cr_builder._config.metadata
             ):
                 config_metadata = self._cr_builder._config.metadata
                 if (
@@ -1096,6 +1100,9 @@ def _deploy_lmeval_cr(self, cr: dict, job_id: str) -> None:
         )
         logger.info("Full Custom Resource being submitted: \n%s", cr_yaml)
 
+        if self._k8s_custom_api is None:
+            raise LMEvalConfigError("Kubernetes custom API not initialized")
+
         try:
             response = self._k8s_custom_api.create_namespaced_custom_object(
                 group=group,
@@ -1324,6 +1331,10 @@ async def job_status(self, benchmark_id: str, job_id: str) -> dict[str, str] | N
             logger.warning("Job %s not found", job_id)
             return None
 
+        if self._k8s_custom_api is None:
+            logger.warning("Kubernetes custom API not initialized")
+            return {"job_id": job_id, "status": JobStatus.scheduled}
+
         try:
             job_metadata = self._job_metadata.get(job_id, {})
             k8s_name = job_metadata.get("k8s_name")
@@ -1392,6 +1403,10 @@ async def job_cancel(self, benchmark_id: str, job_id: str) -> None:
             logger.warning("Job %s not found", job_id)
             return
 
+        if self._k8s_custom_api is None:
+            logger.warning("Kubernetes custom API not initialized")
+            return
+
         try:
             job_metadata = self._job_metadata.get(job_id, {})
             k8s_name = job_metadata.get("k8s_name")
@@ -1455,6 +1470,10 @@ def _get_k8s_cr(self, k8s_name: str) -> dict[str, Any] | None:
         Returns:
             Custom resource as dictionary or None if not found
         """
+        if self._k8s_custom_api is None:
+            logger.warning("Kubernetes custom API not initialized")
+            return None
+
         try:
             group = "trustyai.opendatahub.io"
             version = "v1alpha1"