Merge pull request #101 from socraticDevBlog/20250427-refactordbconnectiondetails

Refactor database connection handling by centralizing configuration i…

Author: socraticDevBlog

backend/app/db_configs.py

@@ -0,0 +1,18 @@
+from dotenv import load_dotenv
+import os
+
+
+def get_db_connection_details():
+    """
+    Load and return database connection details from environment variables.
+    """
+    load_dotenv()
+
+    return {
+        "host": os.getenv("DB_HOST", "localhost"),
+        "port": os.getenv("DB_PORT", "5432"),
+        "user": os.getenv("DB_USER"),
+        "password": os.getenv("DB_PASSWORD"),
+        "database": os.getenv("DB_NAME"),
+        "schema_name": os.getenv("DB_SCHEMA", "public"),
+    }

backend/app/services/db_init.py

@@ -1,21 +1,13 @@
-from dotenv import load_dotenv
-import os
 from app.scripts.postgresql import create_schema_if_not_exists, create_paste_table
+from app.db_configs import get_db_connection_details
 
 
 async def initialize_database():
     """
     Initialize the database by creating the schema and table if they don't exist.
     """
-    load_dotenv()
-    conn_details = {
-        "host": os.getenv("DB_HOST", "localhost"),
-        "port": os.getenv("DB_PORT", "5432"),
-        "user": os.getenv("DB_USER"),
-        "password": os.getenv("DB_PASSWORD"),
-        "database": os.getenv("DB_NAME"),
-    }
-    schema_name = os.getenv("DB_SCHEMA", "public")
+    conn_details = get_db_connection_details()
+    schema_name = conn_details.pop("schema_name")
 
     # Ensure schema and table exist
     await create_schema_if_not_exists(

backend/app/services/dependencies.py

@@ -1,21 +1,11 @@
-from dotenv import load_dotenv
 from app.services.db_client import DBClient
-import os
+from app.db_configs import get_db_connection_details
 
 
 def get_db_client():
     """
     Dependency to provide a DBClient instance.
     """
-
-    load_dotenv()
-
-    conn_details = {
-        "host": os.getenv("DB_HOST", "localhost"),
-        "port": os.getenv("DB_PORT", "5432"),
-        "user": os.getenv("DB_USER"),
-        "password": os.getenv("DB_PASSWORD"),
-        "database": os.getenv("DB_NAME"),
-    }
-    schema_name = os.getenv("DB_SCHEMA", "public")
+    conn_details = get_db_connection_details()
+    schema_name = conn_details.pop("schema_name")
     return DBClient(conn_details, schema_name)

backend/app/utils.py

@@ -1,4 +1,3 @@
-DEFAULT_ENCODING = "utf-8"
 import base64
 
 DEFAULT_ENCODING = "utf-8"
@@ -15,7 +14,6 @@ def is_base64(s: str, encoding: str = DEFAULT_ENCODING) -> bool:
         bool: True if the string is valid Base64, False otherwise.
     """
     try:
-        # Decode and re-encode to verify Base64 validity
         return base64.b64encode(base64.b64decode(s)).decode(encoding=encoding) == s
     except Exception:
         return False