Remove SqlAlchemy dependency from postgres container

Create a new generic database subclass - DependencyFreeDbContainer
Remove test that was testing sqlalchemy support for driver types
Add tests for supported versions of Postgres

Author: logicbomb

core/testcontainers/core/generic.py

@@ -25,6 +25,24 @@
     pass
 
 
+class DependencyFreeDbContainer(DockerContainer):
+    """
+    A generic database without any package dependencies
+    """
+    def start(self) -> 'DbContainer':
+        self._configure()
+        super().start()
+        self._verify_connection()
+        return self
+
+    def _verify_status(self) -> 'DependencyFreeDbContainer':
+        """override this method to ensure the database is running and accepting connections"""
+        raise NotImplementedError
+
+    def _configure(self) -> None:
+        raise NotImplementedError
+
+
 class DbContainer(DockerContainer):
     """
     Generic database container.

postgres/testcontainers/postgres/__init__.py

@@ -11,18 +11,17 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 import os
-import re
-from datetime import UTC, datetime
 from time import sleep
 from typing import Optional
 
 from testcontainers.core.config import MAX_TRIES, SLEEP_TIME
-from testcontainers.core.generic import DbContainer
+from testcontainers.core.generic import DependencyFreeDbContainer
 from testcontainers.core.utils import raise_for_deprecated_parameter
-from testcontainers.core.waiting_utils import wait_container_is_ready
+from testcontainers.core.waiting_utils import (wait_container_is_ready,
+                                               wait_for_logs)
 
 
-class PostgresContainer(DbContainer):
+class PostgresContainer(DependencyFreeDbContainer):
     """
     Postgres database container.
 
@@ -63,33 +62,17 @@ def _configure(self) -> None:
         self.with_env("POSTGRES_PASSWORD", self.password)
         self.with_env("POSTGRES_DB", self.dbname)
 
-    def get_connection_url(self, host=None) -> str:
-        return super()._create_connection_url(
-            dialect=f"postgresql+{self.driver}", username=self.username,
-            password=self.password, dbname=self.dbname, host=host,
-            port=self.port,
-        )
-
     @wait_container_is_ready()
-    def _connect(self) -> None:
-        count = 0
+    def _verify_status(self) -> None:
+        wait_for_logs(self, ".*database system is ready to accept connections.*", MAX_TRIES, SLEEP_TIME)
 
-        # ALTERNATE IMPLEMENTATION based on comments  from @HofmeisterAn
-        # expr = re.compile(r'(?P<ts>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{9}Z).*')
-        # timestamp = datetime.now()
+        count = 0
         while count < MAX_TRIES:
             status, _ = self.exec(f"pg_isready -hlocalhost -p{self.port} -U{self.username}")
             if status == 0:
                 return
 
-            # ALTERNATE IMPLEMENTATION based on comments  from @HofmeisterAn
-            # stdout = self._container.logs(stderr = False, timestamps = True, since = timestamp)
-            # lines = stdout.decode("utf-8").split("\n")
-            # for line in lines:
-            #     if m:= re.match(expr, line):
-            #         timestamp = datetime.fromisoformat(m.groupdict()["ts"]).replace(tzinfo=None)
-            #         if "database system is ready to accept connections" in line:
-            #             return
-
             sleep(SLEEP_TIME)
-            count += 1
+            count += 1
+
+        raise RuntimeError("Postgres could not get into a ready state")

postgres/tests/test_postgres.py

@@ -1,20 +1,15 @@
-import sqlalchemy
 from testcontainers.postgres import PostgresContainer
 
 
 def test_docker_run_postgres():
-    postgres_container = PostgresContainer("postgres:9.5")
-    with postgres_container as postgres:
-        engine = sqlalchemy.create_engine(postgres.get_connection_url())
-        with engine.begin() as connection:
-            result = connection.execute(sqlalchemy.text("select version()"))
-            for row in result:
-                assert row[0].lower().startswith("postgresql 9.5")
+    # https://www.postgresql.org/support/versioning/
+    supported_versions = ["11", "12", "13", "14", "latest"]
 
+    for version in supported_versions:
+        postgres_container = PostgresContainer(f"postgres:{version}")
+        with postgres_container as postgres:
+            status, msg = postgres.exec(f"pg_isready -hlocalhost -p{postgres.port} -U{postgres.username}")
+
+            assert msg.decode("utf-8").endswith("accepting connections\n")
+            assert status == 0
 
-def test_docker_run_postgres_with_driver_pg8000():
-    postgres_container = PostgresContainer("postgres:9.5", driver="pg8000")
-    with postgres_container as postgres:
-        engine = sqlalchemy.create_engine(postgres.get_connection_url())
-        with engine.begin() as connection:
-            connection.execute(sqlalchemy.text("select 1=1"))