From 7b1611f65c25e23d82330d919b718b6814145cb8 Mon Sep 17 00:00:00 2001
From: Christopher Lefelhocz <31426558+clefelhocz2@users.noreply.github.com>
Date: Fri, 26 Jan 2024 15:51:35 -0600
Subject: [PATCH] Fix environment variable not accepting lists (#11722)

Co-authored-by: Chris Pickett <chris.pickett@prefect.io>
---
 src/prefect/server/api/server.py |  2 +-
 src/prefect/settings.py          |  8 ++++++++
 tests/test_engine.py             | 12 +++++++-----
 tests/test_settings.py           |  8 ++++++++
 4 files changed, 24 insertions(+), 6 deletions(-)

diff --git a/src/prefect/server/api/server.py b/src/prefect/server/api/server.py
index 7e2519350acd..947c07705efa 100644
--- a/src/prefect/server/api/server.py
+++ b/src/prefect/server/api/server.py
@@ -501,7 +501,7 @@ def create_app(
             and services will be disabled.
     """
     settings = settings or prefect.settings.get_current_settings()
-    cache_key = (settings, ephemeral)
+    cache_key = (settings.hash_key(), ephemeral)
 
     if cache_key in APP_CACHE and not ignore_cache:
         return APP_CACHE[cache_key]
diff --git a/src/prefect/settings.py b/src/prefect/settings.py
index 9400110904bc..8780bd4ea5c0 100644
--- a/src/prefect/settings.py
+++ b/src/prefect/settings.py
@@ -1660,6 +1660,14 @@ def with_obfuscated_secrets(self):
         settings.__fields_set__.intersection_update(self.__fields_set__)
         return settings
 
+    def hash_key(self) -> str:
+        """
+        Return a hash key for the settings object.  This is needed since some
+        settings may be unhashable.  An example is lists.
+        """
+        env_variables = self.to_environment_variables()
+        return str(hash(tuple((key, value) for key, value in env_variables.items())))
+
     def to_environment_variables(
         self, include: Iterable[Setting] = None, exclude_unset: bool = False
     ) -> Dict[str, str]:
diff --git a/tests/test_engine.py b/tests/test_engine.py
index 3263fb3aefff..a2387d99d9aa 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -1728,7 +1728,7 @@ async def test_global_task_retry_delay_seconds(
         self, mock_anyio_sleep, prefect_client, flow_run, result_factory
     ):
         with temporary_settings(
-            updates={PREFECT_TASK_DEFAULT_RETRY_DELAY_SECONDS: "43"}
+            updates={PREFECT_TASK_DEFAULT_RETRY_DELAY_SECONDS: [3, 5, 9]}
         ):
             # the flow run must be running prior to running tasks
             await prefect_client.set_flow_run_state(
@@ -1739,14 +1739,14 @@ async def test_global_task_retry_delay_seconds(
             # Define a task that fails once and then succeeds
             mock = MagicMock()
 
-            @task(retries=1)
+            @task(retries=3)
             def flaky_function():
                 mock()
 
-                if mock.call_count == 2:
+                if mock.call_count == 4:
                     return 1
 
-                raise ValueError("try again, but only once")
+                raise ValueError("try again")
 
             # Create a task run to test
             task_run = await prefect_client.create_task_run(
@@ -1757,7 +1757,7 @@ def flaky_function():
             )
 
             # Actually run the task
-            with mock_anyio_sleep.assert_sleeps_for(43):
+            with mock_anyio_sleep.assert_sleeps_for(17):
                 await orchestrate_task_run(
                     task=flaky_function,
                     task_run=task_run,
@@ -1769,6 +1769,8 @@ def flaky_function():
                     log_prints=False,
                 )
 
+            assert mock_anyio_sleep.await_count == 3
+
     async def test_retry_condition_fn_retries_after_failure(
         self, mock_anyio_sleep, prefect_client, flow_run, result_factory
     ):
diff --git a/tests/test_settings.py b/tests/test_settings.py
index 9611b61c974d..817fbd855869 100644
--- a/tests/test_settings.py
+++ b/tests/test_settings.py
@@ -243,6 +243,14 @@ def test_settings_to_environment_does_not_use_value_callback(self):
         # generating the environment variables
         assert "PREFECT_UI_API_URL" not in settings.to_environment_variables()
 
+    def test_settings_hash_key(self):
+        settings = Settings(PREFECT_TEST_MODE=True)
+        diff_settings = Settings(PREFECT_TEST_MODE=False)
+
+        assert settings.hash_key() == settings.hash_key()
+
+        assert settings.hash_key() != diff_settings.hash_key()
+
     @pytest.mark.parametrize(
         "log_level_setting",
         [