Fix mypy type hint errors, rename loop functions

Fixed several errors found by the type hint linter mypy and
added underscore prefix for the callback functions
_loop_step_end and _loop_end.

Author: timopollmeier

greenbone/scap/cpe/cli/download.py

@@ -424,10 +424,14 @@ async def download(console: Console, error_console: Console) -> None:
             )
 
             if run_time_file:
+                if until:
+                    run_time = until
+                else:
+                    run_time = datetime.now()
                 # ensure directories exist
                 run_time_file.parent.mkdir(parents=True, exist_ok=True)
                 run_time_file.write_text(
-                    f"{until.isoformat()}\n",
+                    f"{run_time.isoformat()}\n",
                     encoding="utf8",  # type: ignore
                 )
                 console.log(f"Wrote run time to {run_time_file.absolute()}.")

greenbone/scap/cpe_match/db/manager.py

@@ -107,7 +107,7 @@ async def add_cpe_match_strings(
             )
             await self._insert_foreign_data(transaction, match_strings)
 
-        self._cpes = []
+        self._cpe_match_strings = []
 
     async def _insert_foreign_data(
         self,
@@ -150,8 +150,8 @@ async def find(
         index: int | None = None,
         last_modification_start_date: datetime | None = None,
         last_modification_end_date: datetime | None = None,
-        published_start_date: datetime | None = None,
-        published_end_date: datetime | None = None,
+        created_start_date: datetime | None = None,
+        created_end_date: datetime | None = None,
     ) -> AsyncIterator[CPEMatchStringDatabaseModel]:
         clauses = []
 
@@ -171,13 +171,13 @@ async def find(
                 CPEMatchStringDatabaseModel.last_modified
                 <= last_modification_end_date
             )
-        if published_start_date:
+        if created_start_date:
             clauses.append(
-                CPEMatchStringDatabaseModel.published >= published_start_date
+                CPEMatchStringDatabaseModel.created >= created_start_date
             )
-        if published_end_date:
+        if created_end_date:
             clauses.append(
-                CPEMatchStringDatabaseModel.published <= published_end_date
+                CPEMatchStringDatabaseModel.created <= created_end_date
             )
 
         statement = (
@@ -221,8 +221,8 @@ async def count(
         match_criteria_id: str | None,
         last_modification_start_date: datetime | None = None,
         last_modification_end_date: datetime | None = None,
-        published_start_date: datetime | None = None,
-        published_end_date: datetime | None = None,
+        created_start_date: datetime | None = None,
+        created_end_date: datetime | None = None,
     ) -> int:
         clauses = []
 
@@ -242,13 +242,13 @@ async def count(
                 CPEMatchStringDatabaseModel.last_modified
                 <= last_modification_end_date
             )
-        if published_start_date:
+        if created_start_date:
             clauses.append(
-                CPEMatchStringDatabaseModel.published >= published_start_date
+                CPEMatchStringDatabaseModel.created >= created_start_date
             )
-        if published_end_date:
+        if created_end_date:
             clauses.append(
-                CPEMatchStringDatabaseModel.published <= published_end_date
+                CPEMatchStringDatabaseModel.created <= created_end_date
             )
 
         statement = select(

greenbone/scap/cpe_match/json.py

@@ -7,6 +7,7 @@
 from dataclasses import asdict, dataclass
 from datetime import datetime
 from pathlib import Path
+from typing import Sequence
 
 from pontos.nvd.models.cpe_match_string import CPEMatchString
 from rich.console import Console
@@ -65,7 +66,7 @@ def __init__(
         storage_path: Path,
         *,
         compress: bool = False,
-        schema_path: Path = None,
+        schema_path: Path | None = None,
         raise_error_on_validation=False,
     ):
         super().__init__(
@@ -88,7 +89,9 @@ def add_match_string(self, match_string: CPEMatchString) -> None:
             MatchStringItem(match_string=match_string)
         )
 
-    def add_match_strings(self, match_strings: list[CPEMatchString]) -> None:
+    def add_match_strings(
+        self, match_strings: Sequence[CPEMatchString]
+    ) -> None:
         for match_string in match_strings:
             self.add_match_string(match_string)
 

greenbone/scap/cpe_match/producer/nvd_api.py

@@ -9,6 +9,8 @@
 from rich.console import Console
 from rich.progress import Progress
 
+from ...cli import DEFAULT_RETRIES
+from ...errors import ScapError
 from ...generic_cli.producer.nvd_api import NvdApiProducer
 from ..cli.processor import CPE_MATCH_TYPE_PLURAL
 
@@ -69,12 +71,12 @@ def __init__(
         error_console: Console,
         progress: Progress,
         *,
+        retry_attempts: int = DEFAULT_RETRIES,
         nvd_api_key: str | None = None,
-        retry_attempts: int = None,
-        request_results: int = None,
+        request_results: int | None = None,
         request_filter_opts: dict = {},
         start_index: int = 0,
-        verbose: int = None,
+        verbose: int | None = None,
     ):
         """
         Constructor for a CPE match string NVD API producer.
@@ -90,19 +92,21 @@ def __init__(
             start_index: index/offset of the first item to request
             verbose: Verbosity level of log messages.
         """
+        self._nvd_api: CPEMatchApi
+
         super().__init__(
             console,
             error_console,
             progress,
-            nvd_api_key=nvd_api_key,
             retry_attempts=retry_attempts,
+            nvd_api_key=nvd_api_key,
             request_results=request_results,
             request_filter_opts=request_filter_opts,
             start_index=start_index,
             verbose=verbose,
         )
 
-    def _create_nvd_api(self, nvd_api_key: str) -> CPEMatchApi:
+    def _create_nvd_api(self, nvd_api_key: str | None) -> CPEMatchApi:
         """
         Callback used by the constructor to create the NVD API object
         that can be queried for CPE match strings.
@@ -123,6 +127,9 @@ async def _create_nvd_results(self) -> NVDResults[CPEMatchString]:
 
         Returns: The new `NVDResults` object.
         """
+        if self._queue is None:
+            raise ScapError("No queue has been assigned")
+
         return await self._nvd_api.cpe_matches(
             last_modified_start_date=self._request_filter_opts.get(
                 "last_modified_start_date"

greenbone/scap/cpe_match/worker/db.py

@@ -92,6 +92,8 @@ def __init__(
             echo_sql: Whether to print SQL statements.
             verbose: Verbosity level of log messages.
         """
+        self._manager: CPEMatchStringDatabaseManager
+
         super().__init__(
             console,
             error_console,

greenbone/scap/cpe_match/worker/json.py

@@ -111,11 +111,11 @@ async def _handle_chunk(self, chunk: Sequence[CPEMatchString]):
         """
         self._json_manager.add_match_strings(chunk)
 
-    async def loop_end(self) -> None:
+    async def _loop_end(self) -> None:
         """
         Callback handling the exiting the main worker loop.
 
         Makes the JSON manager write the document to the file.
         """
         self._json_manager.write()
-        await super().loop_end()
+        await super()._loop_end()

greenbone/scap/cve/cli/download.py

@@ -432,10 +432,14 @@ async def download(console: Console, error_console: Console):
                 )
 
         if run_time_file:
+            if until:
+                run_time = until
+            else:
+                run_time = datetime.now()
             # ensure directories exist
             run_time_file.parent.mkdir(parents=True, exist_ok=True)
             run_time_file.write_text(
-                f"{until.isoformat()}\n",
+                f"{run_time.isoformat()}\n",
                 encoding="utf8",  # type: ignore
             )
             console.log(f"Wrote run time to {run_time_file.absolute()}.")

greenbone/scap/data_utils/json.py

@@ -123,7 +123,7 @@ class JsonManager:
     def __init__(
         self,
         error_console: Console,
-        schema_path: Path = None,
+        schema_path: Path | None = None,
         raise_error_on_validation=False,
     ):
         """

greenbone/scap/generic_cli/processor.py

@@ -118,7 +118,7 @@ def __init__(
         self._worker.set_queue(self._queue)
 
         self._verbose: int = (
-            verbose if not None else self._arg_defaults["verbose"],
+            verbose if verbose is not None else self._arg_defaults["verbose"]
         )
         "Verbosity level of log messages."
 

greenbone/scap/generic_cli/producer/base.py

@@ -4,7 +4,7 @@
 
 from abc import ABC, abstractmethod
 from argparse import ArgumentParser
-from typing import AsyncContextManager, Generic, TypeVar
+from typing import Any, AsyncContextManager, Generic, TypeVar
 
 from rich.console import Console
 from rich.progress import Progress
@@ -30,10 +30,10 @@ class BaseScapProducer(Generic[T], AsyncContextManager, ABC):
     e.g. `BaseScapProducer[CPE]` will be a producer handling CPE objects.
     """
 
-    _item_type_plural = "SCAP items"
+    _item_type_plural: str = "SCAP items"
     "Plural form of the type of items to use in log messages"
 
-    _arg_defaults = {
+    _arg_defaults: dict[str, Any] = {
         "verbose": DEFAULT_VERBOSITY,
     }
     "Default values for optional arguments."
@@ -79,7 +79,7 @@ def __init__(
         self._verbose = verbose if not None else self._arg_defaults["verbose"]
         "Verbosity level of log messages."
 
-        self._queue: ScapChunkQueue[T] | None = None
+        self._queue: ScapChunkQueue[T]
         "Queue chunks of SCAP items are added to."
 
     @abstractmethod
@@ -110,7 +110,7 @@ async def run_loop(
         It should also create a task for the `progress` object and update it
         regularly.
         """
-        self._queue.set_producer_finished()
+        pass
 
     def set_queue(self, queue: ScapChunkQueue[T]) -> None:
         """