Use pg_isready to check for postgres ready state

Author: logicbomb

postgres/testcontainers/postgres/__init__.py

@@ -11,9 +11,15 @@
 #    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.utils import raise_for_deprecated_parameter
+from testcontainers.core.waiting_utils import wait_container_is_ready
 
 
 class PostgresContainer(DbContainer):
@@ -63,3 +69,27 @@ def get_connection_url(self, host=None) -> str:
             password=self.password, dbname=self.dbname, host=host,
             port=self.port,
         )
+
+    @wait_container_is_ready()
+    def _connect(self) -> None:
+        count = 0
+
+        # 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()
+        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