Repository: PrefectHQ/prefect
Pull request: setting to disable cache globally
closes https://github.com/PrefectHQ/prefect/issues/18161

Adds a new setting `PREFECT_TASKS_DISABLE_CACHING` that allows users to globally disable task caching, even when tasks have explicit `cache_policy` parameters set.

### Example

```python
from prefect import task
from prefect.cache_policies import TASK_SOURCE

@task(cache_policy=TASK_SOURCE)  # Normally would enable caching
def my_task(x):
    return x + 1

# With PREFECT_TASKS_DISABLE_CACHING =true, caching is disabled regardless
```

### Why this matters

This enables users to conditionally disable caching across their entire codebase without modifying task definitions - useful for scenarios like running at scale (where many task definitions might already have a defined `cache_policy`) where caching overhead isn't desired, or toggling caching behavior between environments.


Problem statement:
Prefect caching activated when setting cache_policy
### Bug summary

When I define a task like

```
@task(name="Extract Task", cache_policy=TASK_SOURCE)
def extract(x):
    logger = get_run_logger()
    logger.info("inside extract task")
    return x + 1
```
caching is automatically activated, even though I do not define persist_result=True or set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

What I want to achieve is that I can control caching via
PREFECT_RESULTS_PERSIST_BY_DEFAULT=true / false

to allow me to globally enable or disable caching - while explicitly not setting persists_result in the decorator.
This is currently not possible as a cache_policy setting activates the caching implicitly.

### Version info

```Text
Version:             3.3.7
API version:         0.8.4
Python version:      3.10.13
Git commit:          8f86aaee
Built:               Mon, Apr 28, 2025 03:04 PM
OS/Arch:             win32/AMD64
Profile:             ephemeral
Server type:         ephemeral
Pydantic version:    2.10.6
Server:
  Database:          sqlite
  SQLite version:    3.41.2
```

### Additional context

_No response_

Hints from issue discussion:
hi @OliverKleinBST - thanks for the issue! currently, explicitly setting a `cache_policy` is meant to enable caching.


> What I want to achieve is that I can control caching via
PREFECT_RESULTS_PERSIST_BY_DEFAULT=true / false

> to allow me to globally enable or disable caching - while explicitly not setting persists_result in the decorator.
This is currently not possible as a cache_policy setting activates the caching implicitly.


It is possible to globally enable/disable caching with an env var (`PREFECT_RESULTS_PERSIST_BY_DEFAULT`) without setting `persist_result` in the decorator, you just cannot also explicitly set a `cache_policy`

if you wanted to do something custom based on another setting or env var, you could use a `CacheKeyFnPolicy` that reads your desired flag at runtime, but generally that's not needed, so feel free to share your motivation here

Hi @zzstoatzz 

My motivation is that I need a cache_policy when caching is enabled to exclude one parameter that is not serializable.
`Like in mytask = task(cache_policy=DEFAULT - "param")(function_name)`

This is done when I e.g. run the flow based on some event where the cache can get of relevance.

However, i also run the flow in a automation framework at scale (like hundreds of variations), where I dont need or want caching.
And my thought was I could just disable it with the environment variable assuming that the cache policy setting would then just ignored.

Alternatively of course I could probably create the task definition at runtime based on the env var

```
if cache:
   simulate_task = task(cache_policy=DEFAULT - "logger")(simulate)
else:
   simulate_task = task()(simulate)

```
which felt redundant to me.

hi @OliverKleinBST - that makes sense. I think we can introduce a `PREFECT_TASKS_DISABLE_CACHE` setting that will unequivocally disable caching, regardless of selected `cache_policy` - how does that sound?
sounds great

Commit message: add to list of settings

Patch to review:
diff --git a/src/prefect/settings/models/tasks.py b/src/prefect/settings/models/tasks.py
index 5dd2cef4fb24..6d006c517393 100644
--- a/src/prefect/settings/models/tasks.py
+++ b/src/prefect/settings/models/tasks.py
@@ -63,6 +63,11 @@ class TasksSettings(PrefectBaseSettings):
         description="If `True`, sets the default cache policy on all tasks to `NO_CACHE`.",
     )
 
+    disable_cache: bool = Field(
+        default=False,
+        description="If `True`, disables caching on all tasks regardless of cache policy.",
+    )
+
     default_retries: int = Field(
         default=0,
         ge=0,
diff --git a/src/prefect/tasks.py b/src/prefect/tasks.py
index 0c2829f68180..1e5b816fd389 100644
--- a/src/prefect/tasks.py
+++ b/src/prefect/tasks.py
@@ -466,6 +466,10 @@ def __init__(
             ):
                 persist_result = True
 
+        # Check for global cache disable setting
+        if settings.tasks.disable_cache:
+            persist_result = False
+
         if persist_result is False:
             self.cache_policy = None if cache_policy is None else NO_CACHE
             if cache_policy and cache_policy is not NotSet and cache_policy != NO_CACHE:
diff --git a/tests/test_settings.py b/tests/test_settings.py
index 3923c975fb57..7c21736529cb 100644
--- a/tests/test_settings.py
+++ b/tests/test_settings.py
@@ -435,6 +435,7 @@
     "PREFECT_TASKS_DEFAULT_PERSIST_RESULT": {"test_value": True},
     "PREFECT_TASKS_DEFAULT_RETRIES": {"test_value": 10},
     "PREFECT_TASKS_DEFAULT_RETRY_DELAY_SECONDS": {"test_value": 10},
+    "PREFECT_TASKS_DISABLE_CACHE": {"test_value": False},
     "PREFECT_TASKS_REFRESH_CACHE": {"test_value": True},
     "PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS": {"test_value": 5},
     "PREFECT_TASKS_SCHEDULING_DEFAULT_STORAGE_BLOCK": {"test_value": "block"},
diff --git a/tests/test_tasks.py b/tests/test_tasks.py
index 575f439707e0..a6f193216403 100644
--- a/tests/test_tasks.py
+++ b/tests/test_tasks.py
@@ -2172,6 +2172,50 @@ def test_flow():
         assert await s5.result() == 42
         assert await s6.result() == 42
 
+    async def test_disable_cache_setting_disables_caching_regardless_of_cache_policy(
+        self, caplog
+    ):
+        from prefect.settings import PREFECT_TASKS_DISABLE_CACHE
+
+        with temporary_settings({PREFECT_TASKS_DISABLE_CACHE: True}):
+
+            @task(cache_policy=TASK_SOURCE)
+            def foo(x):
+                return x
+
+        assert foo.cache_policy == NO_CACHE
+        assert foo.persist_result is False
+        assert (
+            "Ignoring `cache_policy` because `persist_result` is False" in caplog.text
+        )
+
+    async def test_disable_cache_setting_allows_normal_caching_when_false(self):
+        from prefect.settings import PREFECT_TASKS_DISABLE_CACHE
+
+        with temporary_settings({PREFECT_TASKS_DISABLE_CACHE: False}):
+
+            @task(cache_policy=TASK_SOURCE)
+            def foo(x):
+                return x
+
+        assert foo.cache_policy == TASK_SOURCE
+        assert foo.persist_result is True
+
+    @pytest.mark.parametrize("cache_policy", [NO_CACHE, None])
+    async def test_does_not_warn_when_false_persist_result_and_none_cache_policy(
+        self, caplog, cache_policy
+    ):
+        @task(persist_result=False, cache_policy=cache_policy)
+        def foo(x):
+            return x
+
+        assert foo.cache_policy == cache_policy
+
+        assert (
+            "Ignoring `cache_policy` because `persist_result` is False"
+            not in caplog.text
+        )
+
 
 class TestCacheFunctionBuiltins:
     async def test_task_input_hash_within_flows(


Code review comments:
diff_hunk: @@ -466,6 +466,10 @@ def __init__(
             ):
                 persist_result = True
 
+        # Check for global cache disable setting
+        if settings.tasks.disable_cache:
+            persist_result = False
path: src/prefect/tasks.py
text: @user1:
I think this could lead to surprising edge cases; was there a reason you chose to do this vs. a more targeted

```python
cache_policy = NO_CACHE
```

?

@author:
good point! 

@author:
updated in [cc11410](https://github.com/PrefectHQ/prefect/pull/18195/commits/cc1141024a01d51c235810bd424af282e4820248)

diff_hunk: @@ -63,6 +63,11 @@ class TasksSettings(PrefectBaseSettings):
         description="If `True`, sets the default cache policy on all tasks to `NO_CACHE`.",
     )
 
+    disable_cache: bool = Field(
path: src/prefect/settings/models/tasks.py
text: @user1:
i'm not suggesting we need to change, but i'm wondering if `disable_caching` is more in line with how we talk about the feature vs. `disable_cache`? wdyt

@author:
yea i kinda like that better tbh