improve type hints (#2966)

## Changes
Add type hints

### Linked issues
None

### Functionality
None

### Tests
- [x] ran unit tests

---------

Co-authored-by: Eric Vergnaud <[email protected]>

Author: ericvergnaud

src/databricks/labs/ucx/assessment/azure.py

@@ -73,7 +73,7 @@ def _get_relevant_service_principals(self) -> list[AzureServicePrincipalInfo]:
 
         # list all relevant service principals in jobs
         all_jobs = list(self._ws.jobs.list(expand_tasks=True))
-        all_clusters_by_id = {c.cluster_id: c for c in self._ws.clusters.list()}
+        all_clusters_by_id = {c.cluster_id: c for c in self._ws.clusters.list() if c.cluster_id}
         for _job, cluster_config in self._get_cluster_configs_from_all_jobs(all_jobs, all_clusters_by_id):
             set_service_principals.update(self._get_azure_spn_from_cluster_config(cluster_config))
 

src/databricks/labs/ucx/assessment/jobs.py

@@ -8,7 +8,7 @@
 from databricks.labs.lsql.backends import SqlBackend
 from databricks.sdk import WorkspaceClient
 from databricks.sdk.service import compute
-from databricks.sdk.service.compute import ClusterDetails
+from databricks.sdk.service.compute import ClusterDetails, ClusterSpec
 from databricks.sdk.service.jobs import (
     BaseJob,
     BaseRun,
@@ -43,7 +43,7 @@ class JobInfo:
 
 class JobsMixin:
     @classmethod
-    def _get_cluster_configs_from_all_jobs(cls, all_jobs, all_clusters_by_id):
+    def _get_cluster_configs_from_all_jobs(cls, all_jobs: list[BaseJob], all_clusters_by_id: dict[str, ClusterDetails]):
         for job in all_jobs:
             if job.settings is None:
                 continue
@@ -54,7 +54,11 @@ def _get_cluster_configs_from_all_jobs(cls, all_jobs, all_clusters_by_id):
             yield from cls._task_clusters(job, all_clusters_by_id)
 
     @classmethod
-    def _task_clusters(cls, job, all_clusters_by_id):
+    def _task_clusters(
+        cls, job: BaseJob, all_clusters_by_id: dict[str, ClusterDetails]
+    ) -> Iterable[tuple[BaseJob, ClusterDetails | ClusterSpec]]:
+        if not job.settings or not job.settings.tasks:
+            return
         for task in job.settings.tasks:
             if task.existing_cluster_id is not None:
                 interactive_cluster = all_clusters_by_id.get(task.existing_cluster_id, None)
@@ -65,7 +69,9 @@ def _task_clusters(cls, job, all_clusters_by_id):
                 yield job, task.new_cluster
 
     @staticmethod
-    def _job_clusters(job):
+    def _job_clusters(job: BaseJob) -> Iterable[tuple[BaseJob, ClusterSpec]]:
+        if not job.settings or not job.settings.job_clusters:
+            return
         for job_cluster in job.settings.job_clusters:
             if job_cluster.new_cluster is None:
                 continue
@@ -79,7 +85,7 @@ def __init__(self, ws: WorkspaceClient, sbe: SqlBackend, schema):
 
     def _crawl(self) -> Iterable[JobInfo]:
         all_jobs = list(self._ws.jobs.list(expand_tasks=True))
-        all_clusters = {c.cluster_id: c for c in self._ws.clusters.list()}
+        all_clusters = {c.cluster_id: c for c in self._ws.clusters.list() if c.cluster_id}
         return self._assess_jobs(all_jobs, all_clusters)
 
     def _assess_jobs(self, all_jobs: list[BaseJob], all_clusters_by_id) -> Iterable[JobInfo]: