[TECH DEBT]: Deal with some nits in the existing crawler code (#2604)

## Changes

This PR deals with some nits that I noticed amongst the crawlers while
working on #2566, but omitted to avoid distracting from the focus of
that PR. These do not result in any functional changes.

Author: asnare

src/databricks/labs/ucx/hive_metastore/grants.py

@@ -245,10 +245,12 @@ def _crawl(self) -> Iterable[Grant]:
         list[Grant]: A list of Grant objects representing the grants found in hive_metastore.
         """
         catalog = "hive_metastore"
-        tasks = [partial(self.grants, catalog=catalog)]
-        # Scanning ANY FILE and ANONYMOUS FUNCTION grants
-        tasks.append(partial(self.grants, catalog=catalog, any_file=True))
-        tasks.append(partial(self.grants, catalog=catalog, anonymous_function=True))
+        tasks = [
+            partial(self.grants, catalog=catalog),
+            # Scanning ANY FILE and ANONYMOUS FUNCTION grants
+            partial(self.grants, catalog=catalog, any_file=True),
+            partial(self.grants, catalog=catalog, anonymous_function=True),
+        ]
         if not self._include_databases:
             # scan all databases, even empty ones
             for row in self._fetch(f"SHOW DATABASES FROM {escape_sql_identifier(catalog)}"):

src/databricks/labs/ucx/hive_metastore/locations.py

@@ -304,7 +304,8 @@ def __init__(self, backend: SqlBackend, ws: WorkspaceClient, inventory_database:
         super().__init__(backend, "hive_metastore", inventory_database, "mounts", Mount)
         self._dbutils = ws.dbutils
 
-    def _deduplicate_mounts(self, mounts: list) -> list:
+    @staticmethod
+    def _deduplicate_mounts(mounts: list) -> list:
         seen = set()
         deduplicated_mounts = []
         for obj in mounts:
@@ -395,7 +396,8 @@ def _try_fetch(self) -> Iterable[Table]:
         ):
             yield Table(*row)
 
-    def _get_tables_paths_from_assessment(self, loaded_records: Iterable[Table]) -> dict[str, str]:
+    @staticmethod
+    def _get_tables_paths_from_assessment(loaded_records: Iterable[Table]) -> dict[str, str]:
         seen = {}
         for rec in loaded_records:
             if not rec.location:
@@ -410,8 +412,8 @@ def _crawl_tables(self, table_paths_from_assessment: dict[str, str]) -> list[Tab
             if self._include_mounts and mount.name not in self._include_mounts:
                 logger.info(f"Filtering mount {mount.name}")
                 continue
-            table_paths = {}
             if self._include_paths_in_mount:
+                table_paths = {}
                 for path in self._include_paths_in_mount:
                     table_paths.update(self._find_delta_log_folders(path))
             else:
@@ -446,7 +448,8 @@ def _crawl_tables(self, table_paths_from_assessment: dict[str, str]) -> list[Tab
         logger.info(f"Found a total of {len(all_tables)} tables in mount points")
         return all_tables
 
-    def _get_table_location(self, mount: Mount, path: str):
+    @staticmethod
+    def _get_table_location(mount: Mount, path: str) -> str:
         """
         There can be different cases for mounts:
             - Mount(name='/mnt/things/a', source='abfss://[email protected]/a')
@@ -457,7 +460,11 @@ def _get_table_location(self, mount: Mount, path: str):
             return path.replace(f"dbfs:{mount.name}/", mount.source)
         return path.replace(f"dbfs:{mount.name}", mount.source)
 
-    def _find_delta_log_folders(self, root_dir: str, delta_log_folders=None) -> dict:
+    def _find_delta_log_folders(
+        self,
+        root_dir: str,
+        delta_log_folders: dict[str, TableInMount] | None = None,
+    ) -> dict[str, TableInMount]:
         if delta_log_folders is None:
             delta_log_folders = {}
         logger.info(f"Listing {root_dir}")
@@ -519,18 +526,22 @@ def _assess_path(self, file_info: FileInfo) -> TableInMount | None:
             return TableInMount(format="PARQUET", is_partitioned=False)
         return None
 
-    def _is_partitioned(self, file_name: str) -> bool:
+    @staticmethod
+    def _is_partitioned(file_name: str) -> bool:
         return '=' in file_name
 
-    def _is_parquet(self, file_name: str) -> bool:
+    @staticmethod
+    def _is_parquet(file_name: str) -> bool:
         parquet_patterns = {'.parquet'}
         return any(pattern in file_name for pattern in parquet_patterns)
 
-    def _is_csv(self, file_name: str) -> bool:
+    @staticmethod
+    def _is_csv(file_name: str) -> bool:
         csv_patterns = {'.csv'}
         return any(pattern in file_name for pattern in csv_patterns)
 
-    def _is_json(self, file_name: str) -> bool:
+    @staticmethod
+    def _is_json(file_name: str) -> bool:
         json_patterns = {'.json'}
         return any(pattern in file_name for pattern in json_patterns)
 

src/databricks/labs/ucx/hive_metastore/tables.py

@@ -562,7 +562,7 @@ def _describe(self, catalog, database, table) -> Table | None:
     def _crawl(self) -> Iterable[Table]:
         """Crawls and lists tables within the specified catalog and database."""
         tasks = []
-        catalog_tables: Collection[Table] = []
+        catalog_tables: Collection[Table]
         catalog = "hive_metastore"
         databases = self._all_databases()
         for database in databases: