feat: using ConfZ for configuration

icon_service_base/database/config.py

@@ -0,0 +1,37 @@
+from pathlib import Path
+from typing import Literal
+
+from confz import ConfZ, ConfZEnvSource, ConfZFileSource
+from pydantic import AnyUrl, SecretStr
+
+CONFIG_DIR = Path(__file__).parent.parent.parent.resolve() / "config"
+
+
+class OperationMode(ConfZ):  # type: ignore
+    environment: Literal["development"] | Literal["production"] = "development"
+
+    CONFIG_SOURCES = ConfZEnvSource(allow=["ENVIRONMENT"])
+
+
+class PostgreSQLConfig(ConfZ):  # type: ignore
+    host: AnyUrl
+    port: int
+    database: str
+    user: str
+    password: SecretStr
+
+    CONFIG_SOURCES = [
+        ConfZFileSource(f"{CONFIG_DIR}/postgres_{OperationMode().environment}.yaml"),
+        ConfZEnvSource(prefix="POSTGRES_", allow=["user", "password"], file=".env"),
+    ]
+
+
+class InfluxDBConfig(ConfZ):  # type: ignore
+    url: AnyUrl
+    org: str
+    token: SecretStr
+
+    CONFIG_SOURCES = [
+        ConfZFileSource(f"{CONFIG_DIR}/influxdb_config.yaml"),
+        ConfZEnvSource(prefix="INFLUXDB_V2_", allow=["token"], file=".env"),
+    ]

icon_service_base/database/influxdb_connection.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 from types import TracebackType
 
-from dotenv import dotenv_values
 from influxdb_client import (  # type: ignore
     Bucket,
     BucketRetentionRules,
@@ -14,13 +13,7 @@ from influxdb_client.client.write_api import SYNCHRONOUS  # type: ignore
 from influxdb_client.rest import ApiException  # type: ignore
 from loguru import logger
 
-# Load the environment variables from the .env file
-env_vars = dotenv_values()
-
-# Access the individual environment variables
-INFLUXDB_V2_URL = str(env_vars.get("INFLUXDB_V2_URL"))
-INFLUXDB_V2_ORG = env_vars.get("INFLUXDB_V2_ORG")
-INFLUXDB_V2_TOKEN = env_vars.get("INFLUXDB_V2_TOKEN")
+from icon_service_base.database.config import InfluxDBConfig
 
 
 class InfluxDBConnection:
@@ -55,9 +48,9 @@ class InfluxDBConnection:
     """
 
     def __init__(self) -> None:
-        self.url = INFLUXDB_V2_URL
-        self.token = INFLUXDB_V2_TOKEN
-        self.org = INFLUXDB_V2_ORG
+        self.url = InfluxDBConfig().url
+        self.token = str(InfluxDBConfig().token)
+        self.org = InfluxDBConfig().org
         self.client: InfluxDBClient
         self.write_api: WriteApi
         self.buckets_api: BucketsApi | None = None

icon_service_base/database/postgres_connection.py

@@ -7,18 +7,9 @@ from types import TracebackType
 from typing import Any
 
 from dateutil.parser import ParserError, parse  # type: ignore
-from dotenv import dotenv_values
 from sqlmodel import Session, SQLModel, create_engine
 
-# Load the environment variables from the .env file
-env_vars = dotenv_values()
-
-# Access the individual environment variables
-POSTGRESQL_USER = env_vars.get("POSTGRESQL_USER")
-POSTGRESQL_PASSWORD = env_vars.get("POSTGRESQL_PASSWORD")
-POSTGRESQL_HOST = env_vars.get("POSTGRESQL_HOST")
-POSTGRESQL_PORT = env_vars.get("POSTGRESQL_PORT")
-POSTGRESQL_DATABASE = env_vars.get("POSTGRESQL_DATABASE")
+from icon_service_base.database.config import PostgreSQLConfig
 
 
 def json_loads_or_return_input(input_string: str) -> dict[str, Any] | Any:
@@ -98,18 +89,6 @@ def deserialize_json_dict(json_string: str) -> Any:
     return result
 
 
-# the docs state that
-# "You should normally have a single engine object for your whole application
-# and re-use it everywhere."
-database_engine = create_engine(
-    f"postgresql://{POSTGRESQL_USER}:{POSTGRESQL_PASSWORD}@{POSTGRESQL_HOST}:"
-    f"{POSTGRESQL_PORT}/{POSTGRESQL_DATABASE}",
-    echo=False,
-    json_serializer=json_dumps,
-    json_deserializer=deserialize_json_dict,
-)
-
-
 class PostgresDatabaseSession(Session):
     """A class to represent a session with the PostgreSQL database.
 
@@ -149,7 +128,16 @@ class PostgresDatabaseSession(Session):
     def __init__(self) -> None:
         """Initializes a new session bound to the database engine."""
 
-        super().__init__(bind=database_engine)
+        super().__init__(
+            bind=create_engine(
+                f"postgresql://{PostgreSQLConfig().user}:{PostgreSQLConfig().password}@"
+                f"{PostgreSQLConfig().host.host}:{PostgreSQLConfig().port}/"
+                f"{PostgreSQLConfig().database}",
+                echo=False,
+                json_serializer=json_dumps,
+                json_deserializer=deserialize_json_dict,
+            )
+        )
 
     def __enter__(self) -> PostgresDatabaseSession:
         """Begins the runtime context related to the database session."""