Checking for errors in the codebase

app/services/safaricom.py

@@ -419,6 +419,10 @@ def verify_signature(request_body: str, request_signature: str) -> bool:
             return False
 
 
+class SafaricomService(DarajaService):
+    """Backward-compatible alias for Daraja service."""
+
+
 # Singleton instance
 daraja_service = DarajaService()
 

ingestion/daraja_client.py

@@ -49,9 +49,9 @@ class DarajaClient:
 
     def __init__(
         self,
-        consumer_key: str,
-        consumer_secret: str,
-        business_shortcode: str,
+        consumer_key: Optional[str] = None,
+        consumer_secret: Optional[str] = None,
+        business_shortcode: Optional[str] = None,
         passkey: Optional[str] = None,
         callback_url: Optional[str] = None,
         environment: str = "sandbox",
@@ -65,10 +65,22 @@ def __init__(
             business_shortcode: M-Pesa business shortcode
             environment: 'sandbox' or 'production'
         """
-        self.consumer_key = consumer_key
-        self.consumer_secret = consumer_secret
-        self.business_shortcode = business_shortcode
-        self.passkey = passkey
+        self.consumer_key = consumer_key or os.getenv("DARAJA_CONSUMER_KEY") or os.getenv(
+            "DARAJA_KEY", ""
+        )
+        self.consumer_secret = consumer_secret or os.getenv(
+            "DARAJA_CONSUMER_SECRET"
+        ) or os.getenv("DARAJA_SECRET", "")
+        self.business_shortcode = (
+            business_shortcode
+            or os.getenv("MPESA_BUSINESS_SHORTCODE")
+            or os.getenv("DARAJA_BUSINESS_SHORTCODE")
+            or os.getenv("DARAJA_SHORTCODE")
+            or os.getenv("DARAJA_C2B_SHORTCODE")
+            or os.getenv("BUSINESS_SHORTCODE")
+            or ""
+        )
+        self.passkey = passkey or os.getenv("MPESA_PASSKEY") or os.getenv("DARAJA_PASSKEY")
         self.callback_url = callback_url
         self.environment = environment
 
@@ -145,7 +157,7 @@ def get_access_token(self) -> str:
         url = f"{self.base_url}/oauth/v1/generate?grant_type=client_credentials"
 
         try:
-            response = self._session.get(
+            response = requests.get(
                 url, auth=(self.consumer_key, self.consumer_secret), timeout=10
             )
             response.raise_for_status()
@@ -168,6 +180,10 @@ def get_access_token(self) -> str:
             logger.error(f"Failed to get access token: {str(e)}")
             raise
 
+    def authenticate(self) -> Dict[str, str]:
+        """Backward-compatible auth helper."""
+        return {"access_token": self.get_access_token()}
+
     def _stk_password(self, timestamp: str) -> str:
         if not self.passkey:
             raise ValueError(
@@ -218,7 +234,7 @@ def c2b_register_url(
         }
 
         try:
-            response = self._session.post(
+            response = requests.post(
                 url, json=payload, headers=headers, timeout=10
             )
             response.raise_for_status()
@@ -271,7 +287,7 @@ def c2b_simulate(
         }
 
         try:
-            response = self._session.post(
+            response = requests.post(
                 url, json=payload, headers=headers, timeout=10
             )
             response.raise_for_status()
@@ -335,7 +351,7 @@ def initiate_stk_push(
         }
 
         try:
-            response = self._session.post(
+            response = requests.post(
                 url, json=payload, headers=headers, timeout=10
             )
             response.raise_for_status()
@@ -346,6 +362,36 @@ def initiate_stk_push(
             logger.error(f"STK push failed: {str(e)}")
             raise
 
+    def register_c2b_urls(
+        self,
+        validation_url: Optional[str] = None,
+        confirmation_url: Optional[str] = None,
+        response_type: str = "Completed",
+    ) -> Dict[str, Any]:
+        """Backward-compatible alias for URL registration."""
+        return self.c2b_register_url(
+            validation_url=validation_url,
+            confirmation_url=confirmation_url,
+            response_type=response_type,
+        )
+
+    def stk_push(
+        self,
+        phone_number: str,
+        amount: int,
+        account_reference: str = "REF123",
+        transaction_desc: str = "Payment",
+        callback_url: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Backward-compatible alias for STK push initiation."""
+        return self.initiate_stk_push(
+            phone_number=phone_number,
+            amount=amount,
+            account_reference=account_reference,
+            callback_url=callback_url,
+            description=transaction_desc,
+        )
+
 
 if __name__ == "__main__":
     # Example usage

ingestion/db_cache.py

@@ -13,7 +13,6 @@
 import redis
 from typing import Any, Optional, Callable
 from functools import wraps
-from datetime import datetime, timedelta
 
 logger = logging.getLogger(__name__)
 
@@ -231,3 +230,53 @@ def schedule_cache_warmup(interval_seconds: int = 3600):
         """
         logger.info(f"Cache warmup scheduled every {interval_seconds}s")
         # TODO: Implement background task for cache warmup
+
+
+class QueryCache:
+    """Simple query cache with TTL and hit/miss tracking."""
+
+    def __init__(self, ttl: int = 300):
+        self.ttl = ttl
+        self._cache = {}
+        self._hits = 0
+        self._misses = 0
+
+    def _build_key(self, query: str, params: tuple = None) -> str:
+        return f"{query}:{str(params)}"
+
+    def set(self, query: str, params: tuple, result: Any) -> None:
+        key = self._build_key(query, params)
+        self._cache[key] = {"value": result, "expires": time.time() + self.ttl}
+
+    def get(self, query: str, params: tuple):
+        key = self._build_key(query, params)
+        entry = self._cache.get(key)
+
+        if not entry:
+            self._misses += 1
+            return None
+
+        if time.time() >= entry["expires"]:
+            del self._cache[key]
+            self._misses += 1
+            return None
+
+        self._hits += 1
+        return entry["value"]
+
+    def invalidate(self, query: str, params: tuple) -> None:
+        key = self._build_key(query, params)
+        self._cache.pop(key, None)
+
+    def clear(self) -> None:
+        self._cache.clear()
+
+    def get_stats(self) -> dict:
+        total = self._hits + self._misses
+        hit_rate = (self._hits / total) if total else 0
+        return {
+            "hits": self._hits,
+            "misses": self._misses,
+            "hit_rate": hit_rate,
+            "entries": len(self._cache),
+        }

ingestion/db_pool.py

@@ -14,8 +14,6 @@
 import os
 import logging
 import psycopg2
-from psycopg2 import pool
-from typing import Optional
 from ingestion.rds_connection import generate_iam_auth_token, load_environment_variables
 
 logger = logging.getLogger(__name__)
@@ -32,6 +30,13 @@ def __init__(
         min_connections: int = 2,
         max_connections: int = 10,
         use_iam_auth: bool = False,
+        minconn: int = None,
+        maxconn: int = None,
+        host: str = None,
+        port: int = None,
+        database: str = None,
+        user: str = None,
+        password: str = None,
     ):
         """
         Initialize database connection pool.
@@ -41,9 +46,16 @@ def __init__(
             max_connections: Maximum pool size
             use_iam_auth: Use AWS RDS IAM authentication
         """
-        self.min_connections = min_connections
-        self.max_connections = max_connections
+        self.min_connections = minconn if minconn is not None else min_connections
+        self.max_connections = maxconn if maxconn is not None else max_connections
         self.use_iam_auth = use_iam_auth
+        self._explicit_config = {
+            "host": host,
+            "port": port,
+            "database": database,
+            "user": user,
+            "password": password,
+        }
         self._init_pool()
 
     def _init_pool(self):
@@ -56,11 +68,11 @@ def _init_pool(self):
                 logger.info(f"Using AWS RDS IAM authentication for {user}@{host}")
             else:
                 # Local PostgreSQL or standard credentials
-                host = os.environ.get("POSTGRES_HOST", "localhost")
-                port = int(os.environ.get("POSTGRES_PORT", "5432"))
-                database = os.environ.get("POSTGRES_DB", "mpesa_analytics")
-                user = os.environ.get("POSTGRES_USER", "data_engineer")
-                password = os.environ.get("POSTGRES_PASSWORD", "change_me")
+                host = self._explicit_config["host"] or os.environ.get("POSTGRES_HOST", "localhost")
+                port = self._explicit_config["port"] or int(os.environ.get("POSTGRES_PORT", "5432"))
+                database = self._explicit_config["database"] or os.environ.get("POSTGRES_DB", "mpesa_analytics")
+                user = self._explicit_config["user"] or os.environ.get("POSTGRES_USER", "data_engineer")
+                password = self._explicit_config["password"] or os.environ.get("POSTGRES_PASSWORD", "change_me")
                 logger.info(f"Using PostgreSQL connection to {user}@{host}:{port}")
 
             self._pool = psycopg2.pool.SimpleConnectionPool(
@@ -115,6 +127,13 @@ def close_all(self):
             self._pool.closeall()
             logger.info("✓ All connections closed")
 
+    # Backward-compatible method names
+    def return_connection(self, conn):
+        self.release_connection(conn)
+
+    def close(self):
+        self.close_all()
+
     @staticmethod
     def get_instance(
         min_connections: int = 2, max_connections: int = 10, use_iam_auth: bool = False
@@ -155,5 +174,5 @@ def get_pooled_connection(use_iam_auth: bool = False):
             cur.execute('SELECT * FROM transactions')
             cur.close()
     """
-    pool = DatabasePool.get_instance(use_iam_auth=use_iam_auth)
-    return PooledConnection(pool)
+    db_pool = DatabasePool.get_instance(use_iam_auth=use_iam_auth)
+    return PooledConnection(db_pool)