Moves experimental_warn setting to experiments model and removes …

…`async_fetch_state_result` (PrefectHQ#15885)

docs/3.0/develop/settings-ref.mdx

@@ -164,38 +164,6 @@ The URL of the Prefect UI. If not set, the client will attempt to infer it.
 **Supported environment variables**:
 `PREFECT_SILENCE_API_URL_MISCONFIGURATION`
 
-### `experimental_warn`
-If `True`, warn on usage of experimental features.
-
-**Type**: `boolean`
-
-**Default**: `True`
-
-**TOML dotted key path**: `experimental_warn`
-
-**Supported environment variables**:
-`PREFECT_EXPERIMENTAL_WARN`
-
-### `async_fetch_state_result`
-
-        Determines whether `State.result()` fetches results automatically or not.
-        In Prefect 2.6.0, the `State.result()` method was updated to be async
-        to facilitate automatic retrieval of results from storage which means when
-        writing async code you must `await` the call. For backwards compatibility,
-        the result is not retrieved by default for async users. You may opt into this
-        per call by passing  `fetch=True` or toggle this setting to change the behavior
-        globally.
-
-
-**Type**: `boolean`
-
-**Default**: `False`
-
-**TOML dotted key path**: `async_fetch_state_result`
-
-**Supported environment variables**:
-`PREFECT_ASYNC_FETCH_STATE_RESULT`
-
 ---
 ## APISettings
 Settings for interacting with the Prefect API
@@ -475,6 +443,18 @@ The default Docker namespace to use when building images.
 ---
 ## ExperimentsSettings
 Settings for configuring experimental features
+### `warn`
+If `True`, warn on usage of experimental features.
+
+**Type**: `boolean`
+
+**Default**: `True`
+
+**TOML dotted key path**: `experiments.warn`
+
+**Supported environment variables**:
+`PREFECT_EXPERIMENTS_WARN`, `PREFECT_EXPERIMENTAL_WARN`
+
 ### `worker_logging_to_api_enabled`
 Enables the logging of worker logs to Prefect Cloud.
 

schemas/settings.schema.json

@@ -309,6 +309,16 @@
         "ExperimentsSettings": {
             "description": "Settings for configuring experimental features",
             "properties": {
+                "warn": {
+                    "default": true,
+                    "description": "If `True`, warn on usage of experimental features.",
+                    "supported_environment_variables": [
+                        "PREFECT_EXPERIMENTS_WARN",
+                        "PREFECT_EXPERIMENTAL_WARN"
+                    ],
+                    "title": "Warn",
+                    "type": "boolean"
+                },
                 "worker_logging_to_api_enabled": {
                     "default": false,
                     "description": "Enables the logging of worker logs to Prefect Cloud.",
@@ -2217,24 +2227,6 @@
             ],
             "title": "Silence Api Url Misconfiguration",
             "type": "boolean"
-        },
-        "experimental_warn": {
-            "default": true,
-            "description": "If `True`, warn on usage of experimental features.",
-            "supported_environment_variables": [
-                "PREFECT_EXPERIMENTAL_WARN"
-            ],
-            "title": "Experimental Warn",
-            "type": "boolean"
-        },
-        "async_fetch_state_result": {
-            "default": false,
-            "description": "\n        Determines whether `State.result()` fetches results automatically or not.\n        In Prefect 2.6.0, the `State.result()` method was updated to be async\n        to facilitate automatic retrieval of results from storage which means when\n        writing async code you must `await` the call. For backwards compatibility,\n        the result is not retrieved by default for async users. You may opt into this\n        per call by passing  `fetch=True` or toggle this setting to change the behavior\n        globally.\n        ",
-            "supported_environment_variables": [
-                "PREFECT_ASYNC_FETCH_STATE_RESULT"
-            ],
-            "title": "Async Fetch State Result",
-            "type": "boolean"
         }
     },
     "title": "Prefect Settings",

src/integrations/prefect-dask/tests/conftest.py

@@ -4,7 +4,6 @@
 
 import pytest
 
-from prefect.settings import PREFECT_ASYNC_FETCH_STATE_RESULT, temporary_settings
 from prefect.testing.utilities import prefect_test_harness
 
 
@@ -46,9 +45,3 @@ def event_loop(request):
     # Workaround for failures in pytest_asyncio 0.17;
     # see https://github.com/pytest-dev/pytest-asyncio/issues/257
     policy.set_event_loop(loop)
-
-
-@pytest.fixture(autouse=True)
-def fetch_state_result():
-    with temporary_settings(updates={PREFECT_ASYNC_FETCH_STATE_RESULT: True}):
-        yield

src/integrations/prefect-ray/tests/conftest.py

@@ -1,16 +1,9 @@
 import pytest
 
-from prefect.settings import PREFECT_ASYNC_FETCH_STATE_RESULT, temporary_settings
 from prefect.testing.utilities import prefect_test_harness
 
 
 @pytest.fixture(scope="session", autouse=True)
 def prefect_db():
     with prefect_test_harness():
         yield
-
-
-@pytest.fixture(autouse=True)
-def fetch_state_result():
-    with temporary_settings(updates={PREFECT_ASYNC_FETCH_STATE_RESULT: True}):
-        yield

src/integrations/prefect-sqlalchemy/tests/conftest.py

@@ -1,6 +1,5 @@
 import pytest
 
-from prefect.settings import PREFECT_ASYNC_FETCH_STATE_RESULT, temporary_settings
 from prefect.testing.utilities import prefect_test_harness
 
 
@@ -11,9 +10,3 @@ def prefect_db():
     """
     with prefect_test_harness():
         yield
-
-
-@pytest.fixture(autouse=True)
-def fetch_state_result():
-    with temporary_settings(updates={PREFECT_ASYNC_FETCH_STATE_RESULT: True}):
-        yield

src/prefect/client/schemas/objects.py

@@ -32,6 +32,7 @@
 from pydantic_extra_types.pendulum_dt import DateTime
 from typing_extensions import Literal, Self, TypeVar
 
+from prefect._internal.compatibility import deprecated
 from prefect._internal.compatibility.migration import getattr_migration
 from prefect._internal.schemas.bases import ObjectBaseModel, PrefectBaseModel
 from prefect._internal.schemas.fields import CreatedBy, UpdatedBy
@@ -220,10 +221,17 @@ def result(self: "State[R]", raise_on_failure: bool = True) -> R:
     def result(self: "State[R]", raise_on_failure: bool = False) -> Union[R, Exception]:
         ...
 
+    @deprecated.deprecated_parameter(
+        "fetch",
+        when=lambda fetch: fetch is not True,
+        start_date="Oct 2024",
+        end_date="Jan 2025",
+        help="Please ensure you are awaiting the call to `result()` when calling in an async context.",
+    )
     def result(
         self,
         raise_on_failure: bool = True,
-        fetch: Optional[bool] = None,
+        fetch: bool = True,
         retry_result_failure: bool = True,
     ) -> Union[R, Exception]:
         """
@@ -248,22 +256,6 @@ def result(
             The result of the run
 
         Examples:
-            >>> from prefect import flow, task
-            >>> @task
-            >>> def my_task(x):
-            >>>     return x
-
-            Get the result from a task future in a flow
-
-            >>> @flow
-            >>> def my_flow():
-            >>>     future = my_task("hello")
-            >>>     state = future.wait()
-            >>>     result = state.result()
-            >>>     print(result)
-            >>> my_flow()
-            hello
-
             Get the result from a flow state
 
             >>> @flow
@@ -307,7 +299,7 @@ def result(
             >>>     raise ValueError("oh no!")
             >>> my_flow.deploy("my_deployment/my_flow")
             >>> flow_run = run_deployment("my_deployment/my_flow")
-            >>> await flow_run.state.result(raise_on_failure=True, fetch=True) # Raises `ValueError("oh no!")`
+            >>> await flow_run.state.result(raise_on_failure=True) # Raises `ValueError("oh no!")`
         """
         from prefect.states import get_state_result
 

src/prefect/settings/models/experiments.py

@@ -1,4 +1,4 @@
-from pydantic import Field
+from pydantic import AliasChoices, AliasPath, Field
 
 from prefect.settings.base import PrefectBaseSettings, _build_settings_config
 
@@ -10,6 +10,14 @@ class ExperimentsSettings(PrefectBaseSettings):
 
     model_config = _build_settings_config(("experiments",))
 
+    warn: bool = Field(
+        default=True,
+        description="If `True`, warn on usage of experimental features.",
+        validation_alias=AliasChoices(
+            AliasPath("warn"), "prefect_experiments_warn", "prefect_experimental_warn"
+        ),
+    )
+
     worker_logging_to_api_enabled: bool = Field(
         default=False,
         description="Enables the logging of worker logs to Prefect Cloud.",

src/prefect/settings/models/root.py

@@ -149,25 +149,6 @@ class Settings(PrefectBaseSettings):
         """,
     )
 
-    experimental_warn: bool = Field(
-        default=True,
-        description="If `True`, warn on usage of experimental features.",
-    )
-
-    # this setting needs to be removed
-    async_fetch_state_result: bool = Field(
-        default=False,
-        description="""
-        Determines whether `State.result()` fetches results automatically or not.
-        In Prefect 2.6.0, the `State.result()` method was updated to be async
-        to facilitate automatic retrieval of results from storage which means when
-        writing async code you must `await` the call. For backwards compatibility,
-        the result is not retrieved by default for async users. You may opt into this
-        per call by passing  `fetch=True` or toggle this setting to change the behavior
-        globally.
-        """,
-    )
-
     ###########################################################################
     # allow deprecated access to PREFECT_SOME_SETTING_NAME
 

src/prefect/states.py

@@ -12,6 +12,7 @@
 import pendulum
 from typing_extensions import TypeGuard
 
+from prefect._internal.compatibility import deprecated
 from prefect.client.schemas import State as State
 from prefect.client.schemas import StateDetails, StateType
 from prefect.exceptions import (
@@ -32,18 +33,24 @@
     ResultRecordMetadata,
     ResultStore,
 )
-from prefect.settings import PREFECT_ASYNC_FETCH_STATE_RESULT
 from prefect.utilities.annotations import BaseAnnotation
 from prefect.utilities.asyncutils import in_async_main_thread, sync_compatible
 from prefect.utilities.collections import ensure_iterable
 
 logger = get_logger("states")
 
 
+@deprecated.deprecated_parameter(
+    "fetch",
+    when=lambda fetch: fetch is not True,
+    start_date="Oct 2024",
+    end_date="Jan 2025",
+    help="Please ensure you are awaiting the call to `result()` when calling in an async context.",
+)
 def get_state_result(
     state: State[R],
     raise_on_failure: bool = True,
-    fetch: Optional[bool] = None,
+    fetch: bool = True,
     retry_result_failure: bool = True,
 ) -> R:
     """
@@ -52,23 +59,17 @@ def get_state_result(
     See `State.result()`
     """
 
-    if fetch is None and (
-        PREFECT_ASYNC_FETCH_STATE_RESULT or not in_async_main_thread()
-    ):
-        # Fetch defaults to `True` for sync users or async users who have opted in
-        fetch = True
-    if not fetch:
-        if fetch is None and in_async_main_thread():
-            warnings.warn(
-                (
-                    "State.result() was called from an async context but not awaited. "
-                    "This method will be updated to return a coroutine by default in "
-                    "the future. Pass `fetch=True` and `await` the call to get rid of "
-                    "this warning."
-                ),
-                DeprecationWarning,
-                stacklevel=2,
-            )
+    if not fetch and in_async_main_thread():
+        warnings.warn(
+            (
+                "State.result() was called from an async context but not awaited. "
+                "This method will be updated to return a coroutine by default in "
+                "the future. Pass `fetch=True` and `await` the call to get rid of "
+                "this warning."
+            ),
+            DeprecationWarning,
+            stacklevel=2,
+        )
 
         return state.data
     else:

tests/conftest.py

@@ -51,7 +51,6 @@
     PREFECT_API_SERVICES_TASK_RUN_RECORDER_ENABLED,
     PREFECT_API_SERVICES_TRIGGERS_ENABLED,
     PREFECT_API_URL,
-    PREFECT_ASYNC_FETCH_STATE_RESULT,
     PREFECT_CLI_COLORS,
     PREFECT_CLI_WRAP_LINES,
     PREFECT_HOME,
@@ -323,7 +322,6 @@ def pytest_sessionstart(session):
             PREFECT_CLI_COLORS: False,
             PREFECT_CLI_WRAP_LINES: False,
             # Enable future change
-            PREFECT_ASYNC_FETCH_STATE_RESULT: True,
             # Enable debug logging
             PREFECT_LOGGING_LEVEL: "DEBUG",
             PREFECT_LOGGING_INTERNAL_LEVEL: "DEBUG",

tests/results/test_result_fetch.py

@@ -1,21 +1,6 @@
-import pytest
-
 from prefect import flow, task
-from prefect.settings import PREFECT_ASYNC_FETCH_STATE_RESULT, temporary_settings
-
-
-@pytest.fixture(autouse=True)
-def disable_fetch_by_default():
-    """
-    The test suite defaults to the future behavior.
-
-    For these tests, we enable the default user behavior.
-    """
-    with temporary_settings({PREFECT_ASYNC_FETCH_STATE_RESULT: False}):
-        yield
 
 
-@pytest.mark.skip(reason="This test is flaky and needs to be fixed")
 async def test_async_result_warnings_are_not_raised_by_engine():
     # Since most of our tests are run with the opt-in globally enabled, this test
     # covers a bunch of features to cover remaining cases where we may internally
@@ -92,8 +77,7 @@ async def foo():
         return 1
 
     state = await foo(return_state=True)
-    with temporary_settings({PREFECT_ASYNC_FETCH_STATE_RESULT: True}):
-        coro = state.result(fetch=True)
+    coro = state.result()
 
     assert await coro == 1
 

tests/test_settings.py

@@ -178,7 +178,6 @@
     "PREFECT_API_TASK_CACHE_KEY_MAX_LENGTH": {"test_value": 10, "legacy": True},
     "PREFECT_API_TLS_INSECURE_SKIP_VERIFY": {"test_value": True},
     "PREFECT_API_URL": {"test_value": "https://api.prefect.io"},
-    "PREFECT_ASYNC_FETCH_STATE_RESULT": {"test_value": True},
     "PREFECT_CLIENT_CSRF_SUPPORT_ENABLED": {"test_value": True},
     "PREFECT_CLIENT_ENABLE_METRICS": {"test_value": True, "legacy": True},
     "PREFECT_CLIENT_MAX_RETRIES": {"test_value": 3},
@@ -227,7 +226,8 @@
         "legacy": True,
     },
     "PREFECT_EVENTS_WEBSOCKET_BACKFILL_PAGE_SIZE": {"test_value": 10, "legacy": True},
-    "PREFECT_EXPERIMENTAL_WARN": {"test_value": True},
+    "PREFECT_EXPERIMENTAL_WARN": {"test_value": True, "legacy": True},
+    "PREFECT_EXPERIMENTS_WARN": {"test_value": True},
     "PREFECT_EXPERIMENTS_WORKER_LOGGING_TO_API_ENABLED": {"test_value": False},
     "PREFECT_FLOW_DEFAULT_RETRIES": {"test_value": 10, "legacy": True},
     "PREFECT_FLOWS_DEFAULT_RETRIES": {"test_value": 10},