From 763ba2b1e79509d18fbd7f157fdb5225edf1ec25 Mon Sep 17 00:00:00 2001
From: Alexander Streed <desertaxle@users.noreply.github.com>
Date: Mon, 12 Aug 2024 08:28:26 -0500
Subject: [PATCH] Avoid raising errors on database passwords that contain a `$`
 character

---
 src/prefect/settings.py |  6 +++-
 tests/test_settings.py  | 67 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 72 insertions(+), 1 deletion(-)

diff --git a/src/prefect/settings.py b/src/prefect/settings.py
index 90667e12e3ab..1e34b7f0a167 100644
--- a/src/prefect/settings.py
+++ b/src/prefect/settings.py
@@ -345,7 +345,11 @@ def templater(settings, value):
             setting.name: setting.value_from(settings) for setting in upstream_settings
         }
         template = string.Template(str(value))
-        return original_type(template.substitute(template_values))
+        # Note the use of `safe_substitute` to avoid raising an exception if a
+        # template value is missing. In this case, template values will be left
+        # as-is in the string. Using `safe_substitute` prevents us raising when
+        # the DB password contains a `$` character.
+        return original_type(template.safe_substitute(template_values))
 
     return templater
 
diff --git a/tests/test_settings.py b/tests/test_settings.py
index 9929811ce165..aab524b507cc 100644
--- a/tests/test_settings.py
+++ b/tests/test_settings.py
@@ -501,6 +501,73 @@ def test_client_retry_extra_codes_invalid(self, extra_codes):
                 PREFECT_CLIENT_RETRY_EXTRA_CODES.value()
 
 
+class TestDatabaseSettings:
+    def test_database_connection_url_templates_password(self):
+        with temporary_settings(
+            {
+                PREFECT_API_DATABASE_CONNECTION_URL: (
+                    "${PREFECT_API_DATABASE_PASSWORD}/test"
+                ),
+                PREFECT_API_DATABASE_PASSWORD: "password",
+            }
+        ):
+            assert PREFECT_API_DATABASE_CONNECTION_URL.value() == "password/test"
+
+    def test_database_connection_url_templates_null_password(self):
+        # Not exactly beautiful behavior here, but I think it's clear.
+        # In the future, we may want to consider raising if attempting to template
+        # a null value.
+        with temporary_settings(
+            {
+                PREFECT_API_DATABASE_CONNECTION_URL: (
+                    "${PREFECT_API_DATABASE_PASSWORD}/test"
+                )
+            }
+        ):
+            assert PREFECT_API_DATABASE_CONNECTION_URL.value() == "None/test"
+
+    def test_warning_if_database_password_set_without_template_string(self):
+        with pytest.warns(
+            UserWarning,
+            match=(
+                "PREFECT_API_DATABASE_PASSWORD is set but not included in the "
+                "PREFECT_API_DATABASE_CONNECTION_URL. "
+                "The provided password will be ignored."
+            ),
+        ):
+            with temporary_settings(
+                {
+                    PREFECT_API_DATABASE_CONNECTION_URL: "test",
+                    PREFECT_API_DATABASE_PASSWORD: "password",
+                }
+            ):
+                pass
+
+    def test_connection_string_with_dollar_sign(self):
+        """
+        Regression test for https://github.com/PrefectHQ/prefect/issues/11067.
+
+        This test ensures that passwords with dollar signs do not cause issues when
+        templating the connection string.
+        """
+        with temporary_settings(
+            {
+                PREFECT_API_DATABASE_CONNECTION_URL: (
+                    "postgresql+asyncpg://"
+                    "the-user:the-$password@"
+                    "the-database-server.example.com:5432"
+                    "/the-database"
+                ),
+            }
+        ):
+            assert PREFECT_API_DATABASE_CONNECTION_URL.value() == (
+                "postgresql+asyncpg://"
+                "the-user:the-$password@"
+                "the-database-server.example.com:5432"
+                "/the-database"
+            )
+
+
 class TestTemporarySettings:
     def test_temporary_settings(self):
         assert PREFECT_TEST_MODE.value() is True