refactor: fix "Any" type and mypy errors in data_sources/ (#2395)

Author: mjhuff

cve_bin_tool/data_sources/curl_source.py

@@ -8,10 +8,9 @@
 import logging
 import re
 from pathlib import Path
-from typing import Any
 
 import aiohttp
-from bs4 import BeautifulSoup
+from bs4 import BeautifulSoup, NavigableString, ResultSet
 
 from cve_bin_tool.async_utils import FileIO, RateLimiter
 from cve_bin_tool.data_sources import (
@@ -85,9 +84,9 @@ async def download_curl_version(self, session: RateLimiter, version: str) -> Non
             html = await response.text()
         soup = BeautifulSoup(html, "html.parser")
         table = soup.find("table")
-        if not table:
+        if not table or isinstance(table, NavigableString):
             return
-        headers = table.find_all("th")
+        headers: ResultSet | list = table.find_all("th")
         headers = list(map(lambda x: x.text.strip().lower(), headers))
         self.LOGGER.debug(headers)
         rows = table.find_all("tr")
@@ -103,7 +102,7 @@ async def download_curl_version(self, session: RateLimiter, version: str) -> Non
         async with FileIO(filepath, "w") as f:
             await f.write(json.dumps(json_data, indent=4))
 
-    def load_curl_version(self, version: str) -> dict[str, Any]:
+    def load_curl_version(self, version: str) -> list[dict[str, str]]:
         """
         Return the dict of CVE data for the given curl version.
         """

cve_bin_tool/data_sources/nvd_source.py

@@ -13,7 +13,6 @@
 import re
 import sqlite3
 from pathlib import Path
-from typing import Any
 
 import aiohttp
 from rich.progress import track
@@ -77,7 +76,7 @@ def __init__(
         self.cve_count = -1
         self.nvd_type = nvd_type
         self.incremental_update = incremental_update
-        self.all_cve_entries: list[dict[str, Any]] | None = None
+        self.all_cve_entries: list[dict[str, object]] | None = None
 
         # store the nvd api key for use later
         self.nvd_api_key = nvd_api_key
@@ -170,7 +169,7 @@ def format_data(self, all_cve_entries):
 
         return cve_data, affects_data
 
-    def parse_node(self, node: dict[str, Any]) -> list[dict[str, str]]:
+    def parse_node(self, node: dict[str, list[dict[str, str]]]) -> list[dict[str, str]]:
         affects_list = []
         if "cpe_match" in node:
             for cpe_match in node["cpe_match"]:
@@ -280,7 +279,9 @@ def format_data_api2(self, all_cve_entries):
 
         return cve_data, affects_data
 
-    def parse_node_api2(self, node: dict[str, Any]) -> list[dict[str, str]]:
+    def parse_node_api2(
+        self, node: dict[str, list[dict[str, str]]]
+    ) -> list[dict[str, str]]:
         affects_list = []
         if "cpeMatch" in node:
             for cpe_match in node["cpeMatch"]:
@@ -478,7 +479,7 @@ async def cache_update(
             with ErrorHandler(mode=self.error_mode, logger=self.LOGGER):
                 raise SHAMismatch(f"{url} (have: {gotsha}, want: {sha})")
 
-    def load_nvd_year(self, year: int) -> dict[str, Any]:
+    def load_nvd_year(self, year: int) -> dict[str, str | object]:
         """
         Return the dict of CVE data for the given year.
         """