Make a solution to getting conninfo as a string

Author: AlanCoding

ansible_base/lib/utils/db.py

@@ -3,9 +3,12 @@
 from typing import Union
 from zlib import crc32
 
+import psycopg
+
 from django.conf import settings
 from django.db import DEFAULT_DB_ALIAS, connection, connections, transaction
 from django.db.backends.postgresql.base import DatabaseWrapper as PsycopgDatabaseWrapper
+from django.db.backends.postgresql.client import DatabaseClient as PsycopgDatabaseClient
 from django.db.migrations.executor import MigrationExecutor
 
 
@@ -167,6 +170,13 @@ def psycopg_kwargs_from_settings_dict(settings_dict: dj_db_dict) -> dict:
     return PsycopgDatabaseWrapper(settings_dict).get_connection_params()
 
 
+def psycopg_conn_string_from_settings_dict(settings_dict: dj_db_dict) -> str:
+    conn_params = psycopg_kwargs_from_settings_dict(settings_dict)
+    conn_params.pop('cursor_factory')
+    conn_params.pop('context')
+    return psycopg.conninfo.make_conninfo(**conn_params)
+
+
 def combine_settings_dict(settings_dict1: dj_db_dict, settings_dict2: dj_db_dict, **extra_options) -> dj_db_dict:
     """Given two Django settings dictionaries, combine them and return a new settings_dict"""
     settings_dict = deepcopy(settings_dict1)

test_app/tests/lib/utils/test_db.py

@@ -1,12 +1,21 @@
 import threading
 import time
 
+import psycopg
+
 import pytest
 from django.db import connection
 from django.db.utils import OperationalError
 from django.test import override_settings
+from django.conf import settings
 
-from ansible_base.lib.utils.db import advisory_lock, get_pg_notify_params, migrations_are_complete, psycopg_kwargs_from_settings_dict
+from ansible_base.lib.utils.db import (
+    advisory_lock,
+    get_pg_notify_params,
+    migrations_are_complete,
+    psycopg_kwargs_from_settings_dict,
+    psycopg_conn_string_from_settings_dict
+)
 
 
 @pytest.mark.django_db
@@ -81,6 +90,30 @@ def test_psycopg_kwargs_from_settings_dict(self):
             psycopg_params = self._trim_python_objects(psycopg_kwargs_from_settings_dict(test_dict))
             assert psycopg_params == self.PSYCOPG_KWARGS
 
+    def test_listener_string_production(self):
+        "This is a test to correspond to PG_NOTIFY_DSN_SERVER type settings in eda-server"
+        args = psycopg_conn_string_from_settings_dict({
+            "ENGINE": "django.db.backends.postgresql",
+            "HOST": "127.0.0.1",
+            "PORT": 5432,
+            "USER": "postgres",
+            "PASSWORD": "DB_PASSWORD",
+            "NAME": "eda",
+            "TIME_ZONE": settings.DATABASES['default']['TIME_ZONE'],
+            "OPTIONS": {
+                "sslmode": "allow",
+                "sslcert": "",
+                "sslkey": "",
+                "sslrootcert": "",
+            },
+        })
+        assert args == "dbname=eda sslmode=allow sslcert='' sslkey='' sslrootcert='' client_encoding=UTF8 user=postgres password=DB_PASSWORD host=127.0.0.1 port=5432"
+
+    def test_listener_string_production_use(self):
+        "This assures that the data we get for the connection string is usable, and demos how to use"
+        args = psycopg_conn_string_from_settings_dict(settings.DATABASES['default'])
+        psycopg.connect(conninfo=args)
+
 
 class TestAdvisoryLock:
     @pytest.fixture(autouse=True)
@@ -104,7 +137,7 @@ def background_task(django_db_blocker):
     def test_determine_lock_is_held(self, django_db_blocker):
         thread = threading.Thread(target=TestAdvisoryLock.background_task, args=(django_db_blocker,))
         thread.start()
-        for _ in range(5):
+        for _ in range(20):
             with advisory_lock('background_task_lock', wait=False) as held:
                 if held is False:
                     break