updates type hints, implements linting suggestions

icon_service_base/database/config.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Literal, Optional
+from typing import Literal
 
 from confz import BaseConfig, EnvSource, FileSource
 from pydantic import AnyUrl, SecretStr
@@ -32,7 +32,7 @@ VENV_DIR = find_dir_upwards(Path(__file__).resolve(), [".venv", "venv"])
 # Look for "deps" directory starting from the current file's directory.
 DEPS_DIR = find_dir_upwards(Path(__file__).resolve(), ["deps"])
 
-CONFIG_DIR: Optional[Path] = None
+CONFIG_DIR: Path | None = None
 
 # If a ".venv" or "venv" directory was found and its parent's name is not the current
 # package name, check for the "database_config" directory inside the parent directory of
@@ -51,7 +51,7 @@ elif DEPS_DIR is not None:
 
 
 class OperationMode(BaseConfig):  # type: ignore
-    environment: Literal["development"] | Literal["production"] = "development"
+    environment: Literal["development", "production"] = "development"
 
     CONFIG_SOURCES = EnvSource(allow=["ENVIRONMENT"])
 

icon_service_base/database/create_config.py

@@ -1,6 +1,6 @@
 import logging
 from pathlib import Path
-from typing import Optional, TypeVar
+from typing import TypeVar
 
 from confz import BaseConfig, FileSource
 
@@ -15,7 +15,7 @@ class NoConfigSourceError(Exception):
 
 def create_config(
     config_class: type[T],
-    config_folder: Optional[Path | str] = None,
+    config_folder: Path | str | None = None,
     config_file: str = "",
 ) -> T:
     if config_class.CONFIG_SOURCES is not None or config_folder is not None:
@@ -23,13 +23,13 @@ def create_config(
         if config_folder is not None:
             config_sources = FileSource(Path(config_folder) / config_file)
         return config_class(config_sources=config_sources)
-    else:
-        error_msg = (
-            "No 'database_config' folder found in the root directory. Please ensure "
-            "that a 'database_config' folder exists in your project's root directory. "
-            "Alternatively, you can provide a different config folder by passing its "
-            "path to the constructor."
-        )
-        logger.error(error_msg)
 
-        raise NoConfigSourceError(error_msg)
+    error_msg = (
+        "No 'database_config' folder found in the root directory. Please ensure "
+        "that a 'database_config' folder exists in your project's root directory. "
+        "Alternatively, you can provide a different config folder by passing its "
+        "path to the constructor."
+    )
+    logger.error(error_msg)
+
+    raise NoConfigSourceError(error_msg)

icon_service_base/database/influxdb_session.py

@@ -1,11 +1,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Iterable
-from dataclasses import dataclass
-from pathlib import Path
-from types import TracebackType
-from typing import Any, NamedTuple, Optional, Union  # noqa: UNT001
+from typing import TYPE_CHECKING, Any, NamedTuple
 
 from influxdb_client import (
     Bucket,
@@ -19,12 +15,19 @@ from influxdb_client import (
 from influxdb_client.client.write.point import DEFAULT_WRITE_PRECISION
 from influxdb_client.client.write_api import SYNCHRONOUS
 from influxdb_client.rest import ApiException
-from reactivex import Observable
 
 from icon_service_base.database.config import InfluxDBConfig
 from icon_service_base.database.create_config import create_config
 
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+    from pathlib import Path
+    from types import TracebackType
+
+    from reactivex import Observable
+
 logger = logging.getLogger(__name__)
+BUCKET_ALREADY_EXISTS = 422
 
 
 class InfluxDBSession:
@@ -60,7 +63,7 @@ class InfluxDBSession:
 
     conf_folder: Path | str
 
-    def __init__(self, config_folder: Optional[Path | str] = None) -> None:
+    def __init__(self, config_folder: Path | str | None = None) -> None:
         self._config = create_config(
             InfluxDBConfig,
             config_folder=config_folder,
@@ -91,20 +94,18 @@ class InfluxDBSession:
     def write(
         self,
         bucket: str,
-        record: Union[  # type: ignore
-            str,
-            Iterable[str],
-            Point,
-            Iterable[Point],
-            dict[str, Any],
-            Iterable[dict[str, Any]],
-            bytes,
-            Iterable[bytes],
-            Observable[Any],
-            NamedTuple,
-            Iterable[NamedTuple],
-        ],
-        org: Optional[str] = None,
+        record: str
+        | Iterable[str]
+        | Point
+        | Iterable[Point]
+        | dict[str, Any]
+        | Iterable[dict[str, Any]]
+        | bytes
+        | Iterable[bytes]
+        | Observable[Any]
+        | NamedTuple
+        | Iterable[NamedTuple],
+        org: str | None = None,
         write_precision: WritePrecision = DEFAULT_WRITE_PRECISION,  # type: ignore
         **kwargs: Any,
     ) -> Any:
@@ -116,7 +117,7 @@ class InfluxDBSession:
             **kwargs,
         )
 
-    def create_bucket(  # noqa: CFQ002
+    def create_bucket(  # noqa: PLR0913
         self,
         bucket: Bucket | None = None,
         bucket_name: str | None = None,
@@ -153,7 +154,7 @@ class InfluxDBSession:
                 org=org,
             )
         except ApiException as e:
-            if e.status == 422:
+            if e.status == BUCKET_ALREADY_EXISTS:
                 logger.debug(e.message)
                 return
             logger.error(e)

icon_service_base/database/postgres_session.py

@@ -4,17 +4,18 @@ import datetime
 import json
 import logging
 import re
-from pathlib import Path
-from types import TracebackType
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any
 
-from confz import FileSource
 from dateutil.parser import ParserError, parse  # type: ignore
 from sqlmodel import Session, SQLModel, create_engine
 
 from icon_service_base.database.config import OperationMode, PostgreSQLConfig
 from icon_service_base.database.create_config import create_config
 
+if TYPE_CHECKING:
+    from pathlib import Path
+    from types import TracebackType
+
 logger = logging.getLogger(__name__)
 
 
@@ -48,7 +49,7 @@ def is_datetime_format(input_string: str) -> bool:
         return False
 
 
-def json_dumps(data: Any) -> str | list:
+def json_dumps(data: Any) -> str | list[Any]:
     """
     Serialize a Python object into a JSON-formatted string, with custom handling for
     datetime and list objects.
@@ -56,7 +57,7 @@ def json_dumps(data: Any) -> str | list:
     # 'Infinity' is an unallowed token in JSON, thus make it a string
     # https://stackoverflow.com/questions/48356938/store-infinity-in-postgres-json-via-django
     pattern = r"(-?Infinity)"
-    result: str | list
+    result: str | list[Any]
 
     if isinstance(data, str):
         if is_datetime_format(data):
@@ -133,7 +134,7 @@ class PostgresDatabaseSession(Session):
 
     conf_folder: Path | str
 
-    def __init__(self, config_folder: Optional[Path | str] = None) -> None:
+    def __init__(self, config_folder: Path | str | None = None) -> None:
         """Initializes a new session bound to the database engine."""
         self._config = create_config(
             PostgreSQLConfig,