use SQLAlchemy built-in password obfuscation

As str(url) will be changing to obfuscate the PW,
use SQLA 1.3 / 1.4 / 2.0 functionality directly.

Change-Id: I4694cc6d2ed7f0463fe0fae8a93ee9ec5df74760
References: https://github.com/sqlalchemy/sqlalchemy/issues/8567

diff --git a/alembic/util/messaging.py b/alembic/util/messaging.py
index dad222f630d7528b7f79e7b0180e91127e95e0f0..54dc04fd2aa11b35027257632882c461c1f75e5f 100644 (file)
--- a/alembic/util/messaging.py
+++ b/alembic/util/messaging.py
@@ -69,12 +69,7 @@ def err(message: str):
 
 def obfuscate_url_pw(input_url: str) -> str:
     u = url.make_url(input_url)
-    if u.password:
-        if sqla_compat.sqla_14:
-            u = u.set(password="XXXXX")
-        else:
-            u.password = "XXXXX"  # type: ignore[misc]
-    return str(u)
+    return sqla_compat.url_render_as_string(u, hide_password=True)
 
 
 def warn(msg: str, stacklevel: int = 2) -> None:

diff --git a/alembic/util/sqla_compat.py b/alembic/util/sqla_compat.py
index 7e98bb5a645827156564b1dc105539f56b6d8a7c..8046c9c46e45cc5591cf63cf57c0ada2b702fd93 100644 (file)
--- a/alembic/util/sqla_compat.py
+++ b/alembic/util/sqla_compat.py
@@ -129,6 +129,13 @@ def _ensure_scope_for_ddl(
             yield
 
 
+def url_render_as_string(url, hide_password=True):
+    if sqla_14:
+        return url.render_as_string(hide_password=hide_password)
+    else:
+        return url.__to_string__(hide_password=hide_password)
+
+
 def _safe_begin_connection_transaction(
     connection: "Connection",
 ) -> "Transaction":

diff --git a/tests/test_command.py b/tests/test_command.py
index af507cbd8e4f01c05d62caf582fffe1cfd58e84c..0c0ce378f3a12c99a2d4063d27a102cf503371c6 100644 (file)
--- a/tests/test_command.py
+++ b/tests/test_command.py
@@ -256,7 +256,7 @@ class CurrentTest(_BufMixin, TestBase):
     def test_current_obfuscate_password(self):
         eq_(
             util.obfuscate_url_pw("postgresql://scott:tiger@localhost/test"),
-            "postgresql://scott:XXXXX@localhost/test",
+            "postgresql://scott:***@localhost/test",
         )
 
     def test_two_heads(self):