feat: Migrate PostgreSQL interactions to SQLAlchemy's async engine, improving connection management, query parameterization, and URI configuration.

app/core/config.py

@@ -4,6 +4,7 @@ Loads environment variables and provides application settings.
 """
 import os
 from typing import Optional, List
+from pydantic import model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -35,10 +36,13 @@ class Settings(BaseSettings):
     POSTGRES_SSL_CERT: Optional[str] = os.getenv("POSTGRES_SSL_CERT")
     POSTGRES_SSL_KEY: Optional[str] = os.getenv("POSTGRES_SSL_KEY")
     
-    @property
-    def POSTGRES_URI(self) -> str:
-        """Build PostgreSQL URI from components"""
-        return f"postgresql+asyncpg://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
+    POSTGRES_URI: Optional[str] = os.getenv("POSTGRES_URI")
+
+    @model_validator(mode='after')
+    def assemble_db_connection(self) -> 'Settings':
+        if not self.POSTGRES_URI:
+            self.POSTGRES_URI = f"postgresql+asyncpg://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
+        return self
 
     # Redis Configuration
     REDIS_HOST: str = os.getenv("REDIS_HOST", "localhost")

app/sql.py

@@ -90,11 +90,7 @@ async def connect_to_database() -> None:
         try:
             async with async_engine.begin() as conn:
                 await conn.execute(text("SELECT 1"))
-            logger.info("Successfully connected to PostgreSQL database", extra={
-                "host": settings.POSTGRES_HOST,
-                "port": settings.POSTGRES_PORT,
-                "database": settings.POSTGRES_DB
-            })
+            logger.info("Successfully connected to PostgreSQL database")
             return
         except Exception as e:
             last_error = e
@@ -104,22 +100,14 @@ async def connect_to_database() -> None:
                 extra={
                     "attempt": attempts,
                     "max_attempts": max_attempts,
-                    "retry_delay_ms": int(delay * 1000),
-                    "host": settings.POSTGRES_HOST,
-                    "port": settings.POSTGRES_PORT,
-                    "database": settings.POSTGRES_DB
+                    "retry_delay_ms": int(delay * 1000)
                 }
             )
             await asyncio.sleep(delay)
             delay = min(delay * settings.POSTGRES_CONNECT_BACKOFF_MULTIPLIER, 30.0)
     logger.error(
         "Failed to connect to PostgreSQL after retries",
-        exc_info=last_error,
-        extra={
-            "host": settings.POSTGRES_HOST,
-            "port": settings.POSTGRES_PORT,
-            "database": settings.POSTGRES_DB
-        }
+        exc_info=last_error
     )
     raise last_error
 

app/sync/catalogues/handler.py

@@ -3,7 +3,8 @@ Catalogue sync handler for MongoDB to PostgreSQL synchronization.
 """
 from typing import Dict, Any, Optional
 from decimal import Decimal, InvalidOperation
-import asyncpg
+from sqlalchemy.ext.asyncio import AsyncConnection
+from sqlalchemy import text
 from motor.motor_asyncio import AsyncIOMotorDatabase
 from datetime import datetime
 from insightfy_utils.logging import get_logger
@@ -252,7 +253,7 @@ class CatalogueSyncHandler(SyncHandler):
     async def upsert_to_postgres(
         self,
         entity: Dict[str, Any],
-        pg_conn: asyncpg.Connection
+        pg_conn: AsyncConnection
     ) -> bool:
         """
         Upsert catalogue to PostgreSQL trans.catalogue_ref table.
@@ -312,21 +313,20 @@ class CatalogueSyncHandler(SyncHandler):
             
             # Build UPSERT query
             columns = list(mapped_entity.keys())
-            placeholders = [f"${i+1}" for i in range(len(columns))]
-            values = [mapped_entity[col] for col in columns]
+            placeholders = [f":{col}" for col in columns]
             
             # Build UPDATE clause (exclude primary key)
             update_columns = [col for col in columns if col != "catalogue_id"]
             update_clause = ", ".join([f"{col} = EXCLUDED.{col}" for col in update_columns])
             
-            query = f"""
+            query = text(f"""
                 INSERT INTO trans.catalogue_ref ({', '.join(columns)})
                 VALUES ({', '.join(placeholders)})
                 ON CONFLICT (catalogue_id)
                 DO UPDATE SET {update_clause}
-            """
+            """)
             
-            await pg_conn.execute(query, *values)
+            await pg_conn.execute(query, mapped_entity)
             
             logger.debug(
                 "Catalogue upserted to PostgreSQL",

app/sync/catalogues/service.py

@@ -12,7 +12,7 @@ from app.sync.catalogues.handler import CatalogueSyncHandler
 from app.sync.common.retry import RetryManager
 from app.sync.common.monitoring import SyncMonitoringService
 from app.sync.common.models import SyncOperation
-from app.postgres import get_postgres_connection, release_postgres_connection
+from app.sql import async_engine
 
 logger = get_logger(__name__)
 
@@ -312,12 +312,7 @@ class CatalogueSyncService:
         Raises:
             Exception: If sync fails
         """
-        pg_conn = None
-        
-        try:
-            # Acquire PostgreSQL connection
-            pg_conn = await get_postgres_connection()
-            
+        async with async_engine.begin() as pg_conn:
             # Perform sync
             success = await self.handler.sync(
                 entity_id=catalogue_id,
@@ -326,11 +321,6 @@ class CatalogueSyncService:
             )
             
             return success
-            
-        finally:
-            # Always release connection
-            if pg_conn:
-                await release_postgres_connection(pg_conn)
     
     def get_queue_size(self) -> int:
         """

app/sync/common/handler.py

@@ -1,9 +1,7 @@
-"""
-Base sync handler with shared sync logic.
-"""
 from abc import ABC, abstractmethod
 from typing import Dict, Any, Optional
-import asyncpg
+from sqlalchemy.ext.asyncio import AsyncConnection
+from sqlalchemy import text
 from motor.motor_asyncio import AsyncIOMotorDatabase
 from datetime import datetime
 from insightfy_utils.logging import get_logger
@@ -86,7 +84,7 @@ class SyncHandler(ABC):
     async def upsert_to_postgres(
         self,
         entity: Dict[str, Any],
-        pg_conn: asyncpg.Connection
+        pg_conn: AsyncConnection
     ) -> bool:
         """
         Upsert entity to PostgreSQL.
@@ -104,7 +102,7 @@ class SyncHandler(ABC):
         self,
         entity_id: str,
         mongo_db: AsyncIOMotorDatabase,
-        pg_conn: asyncpg.Connection
+        pg_conn: AsyncConnection
     ) -> bool:
         """
         Sync entity from MongoDB to PostgreSQL.
@@ -203,7 +201,7 @@ class SyncHandler(ABC):
         self,
         entity_id: str,
         mongo_updated_at: datetime,
-        pg_conn: asyncpg.Connection,
+        pg_conn: AsyncConnection,
         table_name: str,
         id_column: str
     ) -> bool:
@@ -221,8 +219,10 @@ class SyncHandler(ABC):
             True if should update (MongoDB is newer or record doesn't exist), False otherwise
         """
         try:
-            query = f"SELECT updated_at FROM {table_name} WHERE {id_column} = $1"
-            pg_updated_at = await pg_conn.fetchval(query, entity_id)
+            # Replaced $1 with :id and fetchval with execute().scalar()
+            query = text(f"SELECT updated_at FROM {table_name} WHERE {id_column} = :id")
+            result = await pg_conn.execute(query, {"id": entity_id})
+            pg_updated_at = result.scalar()
             
             if pg_updated_at is None:
                 # Record doesn't exist, should insert

app/sync/employees/handler.py

@@ -2,7 +2,8 @@
 Employee sync handler for MongoDB to PostgreSQL synchronization.
 """
 from typing import Dict, Any, Optional
-import asyncpg
+from sqlalchemy.ext.asyncio import AsyncConnection
+from sqlalchemy import text
 from motor.motor_asyncio import AsyncIOMotorDatabase
 from datetime import datetime
 from insightfy_utils.logging import get_logger
@@ -176,7 +177,7 @@ class EmployeeSyncHandler(SyncHandler):
     async def upsert_to_postgres(
         self,
         entity: Dict[str, Any],
-        pg_conn: asyncpg.Connection
+        pg_conn: AsyncConnection
     ) -> bool:
         """
         Upsert employee to PostgreSQL trans.employees_ref table.
@@ -232,21 +233,20 @@ class EmployeeSyncHandler(SyncHandler):
             
             # Build UPSERT query
             columns = list(mapped_entity.keys())
-            placeholders = [f"${i+1}" for i in range(len(columns))]
-            values = [mapped_entity[col] for col in columns]
+            placeholders = [f":{col}" for col in columns]
             
             # Build UPDATE clause (exclude primary key)
             update_columns = [col for col in columns if col != "employee_id"]
             update_clause = ", ".join([f"{col} = EXCLUDED.{col}" for col in update_columns])
             
-            query = f"""
+            query = text(f"""
                 INSERT INTO trans.employees_ref ({', '.join(columns)})
                 VALUES ({', '.join(placeholders)})
                 ON CONFLICT (employee_id)
                 DO UPDATE SET {update_clause}
-            """
+            """)
             
-            await pg_conn.execute(query, *values)
+            await pg_conn.execute(query, mapped_entity)
             
             logger.debug(
                 "Employee upserted to PostgreSQL",

app/sync/employees/service.py

@@ -12,7 +12,7 @@ from app.sync.employees.handler import EmployeeSyncHandler
 from app.sync.common.retry import RetryManager
 from app.sync.common.monitoring import SyncMonitoringService
 from app.sync.common.models import SyncOperation
-from app.postgres import get_postgres_connection, release_postgres_connection
+from app.sql import async_engine
 
 logger = get_logger(__name__)
 
@@ -312,12 +312,7 @@ class EmployeeSyncService:
         Raises:
             Exception: If sync fails
         """
-        pg_conn = None
-        
-        try:
-            # Acquire PostgreSQL connection
-            pg_conn = await get_postgres_connection()
-            
+        async with async_engine.begin() as pg_conn:
             # Perform sync
             success = await self.handler.sync(
                 entity_id=employee_id,
@@ -326,11 +321,6 @@ class EmployeeSyncService:
             )
             
             return success
-            
-        finally:
-            # Always release connection
-            if pg_conn:
-                await release_postgres_connection(pg_conn)
     
     def get_queue_size(self) -> int:
         """

app/sync/merchants/handler.py

@@ -2,7 +2,8 @@
 Merchant sync handler for MongoDB to PostgreSQL synchronization.
 """
 from typing import Dict, Any, Optional
-import asyncpg
+from sqlalchemy.ext.asyncio import AsyncConnection
+from sqlalchemy import text
 from motor.motor_asyncio import AsyncIOMotorDatabase
 from datetime import datetime
 from insightfy_utils.logging import get_logger
@@ -159,7 +160,7 @@ class MerchantSyncHandler(SyncHandler):
     async def upsert_to_postgres(
         self,
         entity: Dict[str, Any],
-        pg_conn: asyncpg.Connection
+        pg_conn: AsyncConnection
     ) -> bool:
         """
         Upsert merchant to PostgreSQL trans.merchants_ref table.
@@ -211,21 +212,20 @@ class MerchantSyncHandler(SyncHandler):
             
             # Build UPSERT query
             columns = list(mapped_entity.keys())
-            placeholders = [f"${i+1}" for i in range(len(columns))]
-            values = [mapped_entity[col] for col in columns]
+            placeholders = [f":{col}" for col in columns]
             
             # Build UPDATE clause (exclude primary key)
             update_columns = [col for col in columns if col != "merchant_id"]
             update_clause = ", ".join([f"{col} = EXCLUDED.{col}" for col in update_columns])
             
-            query = f"""
+            query = text(f"""
                 INSERT INTO trans.merchants_ref ({', '.join(columns)})
                 VALUES ({', '.join(placeholders)})
                 ON CONFLICT (merchant_id)
                 DO UPDATE SET {update_clause}
-            """
+            """)
             
-            await pg_conn.execute(query, *values)
+            await pg_conn.execute(query, mapped_entity)
             
             logger.debug(
                 "Merchant upserted to PostgreSQL",

app/sync/merchants/service.py

@@ -12,7 +12,7 @@ from app.sync.merchants.handler import MerchantSyncHandler
 from app.sync.common.retry import RetryManager
 from app.sync.common.monitoring import SyncMonitoringService
 from app.sync.common.models import SyncOperation
-from app.postgres import get_postgres_connection, release_postgres_connection
+from app.sql import async_engine
 
 logger = get_logger(__name__)
 
@@ -312,12 +312,7 @@ class MerchantSyncService:
         Raises:
             Exception: If sync fails
         """
-        pg_conn = None
-        
-        try:
-            # Acquire PostgreSQL connection
-            pg_conn = await get_postgres_connection()
-            
+        async with async_engine.begin() as pg_conn:
             # Perform sync
             success = await self.handler.sync(
                 entity_id=merchant_id,
@@ -326,11 +321,6 @@ class MerchantSyncService:
             )
             
             return success
-            
-        finally:
-            # Always release connection
-            if pg_conn:
-                await release_postgres_connection(pg_conn)
     
     def get_queue_size(self) -> int:
         """