diff --git a/icon_service_base/database/config.py b/icon_service_base/database/config.py index 99d655b..c30979d 100644 --- a/icon_service_base/database/config.py +++ b/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"]) diff --git a/icon_service_base/database/create_config.py b/icon_service_base/database/create_config.py index 9cacaa3..772bead 100644 --- a/icon_service_base/database/create_config.py +++ b/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) diff --git a/icon_service_base/database/influxdb_session.py b/icon_service_base/database/influxdb_session.py index 96b7949..3b88185 100644 --- a/icon_service_base/database/influxdb_session.py +++ b/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) diff --git a/icon_service_base/database/postgres_session.py b/icon_service_base/database/postgres_session.py index d71f9ee..e96dc2d 100644 --- a/icon_service_base/database/postgres_session.py +++ b/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,