PostgreSQL connection caching and reuse

.github/copilot-instructions.md

@@ -31,6 +31,7 @@ Python style
 Patterns
 - `__init__` methods must not raise exceptions; defer validation and connection to first use (lazy init)
 - Writers: inherit from `Writer(ABC)`, implement `write(topic, message) -> (bool, str|None)` and `check_health() -> (bool, str)`
+- PostgreSQL: `WriterPostgres` and `ReaderPostgres` cache a single connection per instance
 - Route dispatch via `ROUTE_MAP` dict mapping routes to handler functions in `event_gate_lambda.py` and `event_stats_lambda.py`
 - Separate business logic from environment access (env vars, file I/O, network calls)
 - No duplicate validation; centralize parsing in one layer where practical

.github/dependabot.yml

@@ -14,6 +14,10 @@ updates:
     commit-message:
       prefix: "chore"
       include: "scope"
+    groups:
+      github-actions:
+        patterns:
+          - "*"
 
   - package-ecosystem: "pip"
     directory: "/"
@@ -31,3 +35,7 @@ updates:
       include: "scope"
     allow:
       - dependency-type: "direct"
+    groups:
+      python-dependencies:
+        patterns:
+          - "*"

src/readers/reader_postgres.py

@@ -34,6 +34,7 @@
 try:
     import psycopg2
     from psycopg2 import Error as PsycopgError
+    from psycopg2 import OperationalError
     from psycopg2 import sql as psycopg2_sql
 except ImportError:
     psycopg2 = None  # type: ignore
@@ -42,6 +43,9 @@
     class PsycopgError(Exception):  # type: ignore
         """Shim psycopg2 error base when psycopg2 is not installed."""
 
+    class OperationalError(PsycopgError):  # type: ignore
+        """Shim psycopg2 OperationalError when psycopg2 is not installed."""
+
 
 logger = logging.getLogger(__name__)
 
@@ -70,6 +74,7 @@ def __init__(self) -> None:
         self._secret_name = os.environ.get("POSTGRES_SECRET_NAME", "")
         self._secret_region = os.environ.get("POSTGRES_SECRET_REGION", "")
         self._db_config: dict[str, Any] | None = None
+        self._connection: Any | None = None
         logger.debug("Initialized PostgreSQL reader.")
 
     def _load_db_config(self) -> dict[str, Any]:
@@ -81,6 +86,22 @@ def _load_db_config(self) -> dict[str, Any]:
             raise RuntimeError("Failed to load database configuration.")
         return config
 
+    def _get_connection(self) -> Any:
+        """Return a cached database connection, creating one if needed."""
+        if self._connection is not None and not self._connection.closed:
+            return self._connection
+        db_config = self._load_db_config()
+        self._connection = psycopg2.connect(  # type: ignore[attr-defined]
+            database=db_config["database"],
+            host=db_config["host"],
+            user=db_config["user"],
+            password=db_config["password"],
+            port=db_config["port"],
+            options="-c statement_timeout=30000 -c default_transaction_read_only=on",
+        )
+        logger.debug("New PostgreSQL reader connection established.")
+        return self._connection
+
     def read_stats(
         self,
         timestamp_start: int | None = None,
@@ -124,20 +145,23 @@ def read_stats(
         params.append(limit + 1)
 
         try:
-            with psycopg2.connect(  # type: ignore[attr-defined]
-                database=db_config["database"],
-                host=db_config["host"],
-                user=db_config["user"],
-                password=db_config["password"],
-                port=db_config["port"],
-                options="-c statement_timeout=30000 -c default_transaction_read_only=on",
-            ) as connection:
-                with connection.cursor() as db_cursor:
-                    db_cursor.execute(query, params)
-                    col_names = [desc[0] for desc in db_cursor.description]  # type: ignore[union-attr]
-                    raw_rows = db_cursor.fetchall()
+            for attempt in range(2):
+                try:
+                    connection = self._get_connection()
+                    with connection.cursor() as db_cursor:
+                        db_cursor.execute(query, params)
+                        col_names = [desc[0] for desc in db_cursor.description]  # type: ignore[union-attr]
+                        raw_rows = db_cursor.fetchall()
+                    connection.rollback()
+                    break
+                except OperationalError as exc:
+                    self._connection = None
+                    if attempt > 0:
+                        raise RuntimeError(f"Database connection failed after retry: {exc}") from exc
+                    logger.warning("PostgreSQL connection lost, reconnecting.")
         except PsycopgError as exc:
-            raise RuntimeError(f"Database query failed: {exc}") from exc
+            self._connection = None
+            raise RuntimeError(f"Database query error: {exc}") from exc
 
         rows = [dict(zip(col_names, row, strict=True)) for row in raw_rows]
 

src/writers/writer_postgres.py

@@ -31,6 +31,7 @@
 try:
     import psycopg2
     from psycopg2 import Error as PsycopgError
+    from psycopg2 import OperationalError
     from psycopg2 import sql as psycopg2_sql
 except ImportError:
     psycopg2 = None  # type: ignore
@@ -39,6 +40,9 @@
     class PsycopgError(Exception):  # type: ignore
         """Shim psycopg2 error base when psycopg2 is not installed."""
 
+    class OperationalError(PsycopgError):  # type: ignore
+        """Shim psycopg2 OperationalError when psycopg2 is not installed."""
+
 
 logger = logging.getLogger(__name__)
 
@@ -53,6 +57,7 @@ def __init__(self, config: dict[str, Any]) -> None:
         self._secret_name = os.environ.get("POSTGRES_SECRET_NAME", "")
         self._secret_region = os.environ.get("POSTGRES_SECRET_REGION", "")
         self._db_config: dict[str, Any | None] | None = None
+        self._connection: Any | None = None
         logger.debug("Initialized PostgreSQL writer.")
 
     def _load_db_config(self) -> dict[str, Any]:
@@ -61,6 +66,21 @@ def _load_db_config(self) -> dict[str, Any]:
             self._db_config = load_postgres_config(self._secret_name, self._secret_region)
         return self._db_config  # type: ignore[return-value]
 
+    def _get_connection(self) -> Any:
+        """Return a cached database connection, creating one if needed."""
+        if self._connection is not None and not self._connection.closed:
+            return self._connection
+        db_config = self._load_db_config()
+        self._connection = psycopg2.connect(  # type: ignore[attr-defined]
+            database=db_config["database"],
+            host=db_config["host"],
+            user=db_config["user"],
+            password=db_config["password"],
+            port=db_config["port"],
+        )
+        logger.debug("New PostgreSQL writer connection established.")
+        return self._connection
+
     def _postgres_edla_write(self, cursor: Any, table: str, message: dict[str, Any]) -> None:
         """Insert a dlchange style event row.
         Args:
@@ -278,23 +298,25 @@ def write(self, topic_name: str, message: dict[str, Any]) -> tuple[bool, str | N
 
             table_info = TOPIC_TABLE_MAP[topic_name]
 
-            with psycopg2.connect(  # type: ignore[attr-defined]
-                database=db_config["database"],
-                host=db_config["host"],
-                user=db_config["user"],
-                password=db_config["password"],
-                port=db_config["port"],
-            ) as connection:
-                with connection.cursor() as cursor:
-                    if topic_name == "public.cps.za.dlchange":
-                        self._postgres_edla_write(cursor, table_info["main"], message)
-                    elif topic_name == "public.cps.za.runs":
-                        self._postgres_run_write(cursor, table_info["main"], table_info["jobs"], message)
-                    elif topic_name == "public.cps.za.test":
-                        self._postgres_test_write(cursor, table_info["main"], message)
-
-                connection.commit()
+            for attempt in range(2):
+                try:
+                    connection = self._get_connection()
+                    with connection.cursor() as cursor:
+                        if topic_name == "public.cps.za.dlchange":
+                            self._postgres_edla_write(cursor, table_info["main"], message)
+                        elif topic_name == "public.cps.za.runs":
+                            self._postgres_run_write(cursor, table_info["main"], table_info["jobs"], message)
+                        elif topic_name == "public.cps.za.test":
+                            self._postgres_test_write(cursor, table_info["main"], message)
+                    connection.commit()
+                    break
+                except OperationalError:
+                    self._connection = None
+                    if attempt > 0:
+                        raise
+                    logger.warning("PostgreSQL connection lost, reconnecting.")
         except (RuntimeError, PsycopgError, BotoCoreError, ClientError, ValueError, KeyError) as e:
+            self._connection = None
             err_msg = f"The Postgres writer failed with unknown error: {str(e)}"
             logger.exception(err_msg)
             return False, err_msg

tests/integration/test_connection_reuse.py

@@ -0,0 +1,99 @@
+#
+# Copyright 2026 ABSA Group Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+import time
+import uuid
+
+import pytest
+
+from tests.integration.conftest import EventGateTestClient, EventStatsTestClient
+
+
+def _make_test_event() -> dict:
+    """Build a minimal runs event payload."""
+    now_ms = int(time.time() * 1000)
+    return {
+        "event_id": str(uuid.uuid4()),
+        "job_ref": f"conn-reuse-{uuid.uuid4().hex[:8]}",
+        "tenant_id": "CONN_REUSE_TEST",
+        "source_app": "integration-conn-reuse",
+        "source_app_version": "1.0.0",
+        "environment": "test",
+        "timestamp_start": now_ms - 60000,
+        "timestamp_end": now_ms,
+        "jobs": [
+            {
+                "catalog_id": "db.schema.conn_reuse_table",
+                "status": "succeeded",
+                "timestamp_start": now_ms - 60000,
+                "timestamp_end": now_ms,
+            }
+        ],
+    }
+
+
+class TestWriterConnectionReuse:
+    """Verify that WriterPostgres reuses connections across invocations."""
+
+    @pytest.fixture(scope="class", autouse=True)
+    def seed_events(self, eventgate_client: EventGateTestClient, valid_token: str) -> None:
+        """Post events so the writer connection is established."""
+        for _ in range(2):
+            event = _make_test_event()
+            response = eventgate_client.post_event("public.cps.za.runs", event, token=valid_token)
+            assert 202 == response["statusCode"]
+
+    def test_writer_reuses_connection_across_writes(
+        self, seed_events: None, eventgate_client: EventGateTestClient, valid_token: str
+    ) -> None:
+        """Test that subsequent writes reuse the same cached connection."""
+        from src.event_gate_lambda import writers
+
+        writer = writers["postgres"]
+        conn_before = writer._connection
+        assert conn_before is not None
+        assert 0 == conn_before.closed
+
+        event = _make_test_event()
+        response = eventgate_client.post_event("public.cps.za.runs", event, token=valid_token)
+        assert 202 == response["statusCode"]
+        assert conn_before is writer._connection
+
+
+class TestReaderConnectionReuse:
+    """Verify that ReaderPostgres reuses connections across invocations."""
+
+    @pytest.fixture(scope="class", autouse=True)
+    def seed_events(self, eventgate_client: EventGateTestClient, valid_token: str) -> None:
+        """Seed events so stats queries return data."""
+        for _ in range(2):
+            event = _make_test_event()
+            response = eventgate_client.post_event("public.cps.za.runs", event, token=valid_token)
+            assert 202 == response["statusCode"]
+
+    def test_reader_reuses_connection_across_reads(self, seed_events: None, stats_client: EventStatsTestClient) -> None:
+        """Test that successive queries reuse the same cached connection."""
+        from src.event_stats_lambda import reader_postgres
+
+        stats_client.post_stats("public.cps.za.runs", {})
+
+        conn_after_first = reader_postgres._connection
+        assert conn_after_first is not None
+        assert 0 == conn_after_first.closed
+
+        stats_client.post_stats("public.cps.za.runs", {})
+        assert conn_after_first is reader_postgres._connection