Ah, this is a classic case of conflicting settings. The error you're seeing is actually expected behavior because caching and result persistence are interlinked in Prefect. Here's what's happening: 1. Caching requires result persistence to function - it needs somewhere to store the cached results 2. When you set

persist_result=False

, you're telling Prefect not to store any results 3. This creates a conflict when Prefect tries to compute a cache key, as there would be nowhere to store the cached result To resolve this, you have two options: 1. If you want to use caching: - Remove

persist_result=False

- Let Prefect handle result persistence normally 2. If you want to disable result persistence: - Keep

persist_result=False

- Accept that caching won't work - Ignore the warning message as it's just informing you that caching is disabled Here's an example of how to properly set up a flow with caching:

python
from prefect import flow

@flow(cache_key_fn=lambda context: "my-cache-key")  # if you want caching
def my_flow():
    return "some result"

Or if you really don't want result persistence or caching:

python
from prefect import flow

@flow(persist_result=False)
def my_flow():
    return "some result"

You can find more details about caching and result persistence in the caching documentation and results documentation.

Here's how to implement this scenario:

python
from prefect import flow, task
from datetime import timedelta

# This task will not persist results
@task(persist_result=False)
def non_cached_task():
    return "not cached"

# This task will persist results and use caching
@task(
    persist_result=True,  # Required for caching
    cache_key_fn=lambda context: "my-cached-task",  # Or any other cache key logic
    cache_expiration=timedelta(hours=1)  # Optional: set how long to keep the cache
)
def cached_task():
    return "cached result"

# The flow defaults to not persisting results
@flow(persist_result=False)
def my_flow():
    # This result won't be persisted
    non_cached_task()
    
    # This result will be persisted and cached
    cached_task()

Key points: 1. Set

persist_result=False

on the flow to make it the default for all tasks 2. For the specific task you want to cache: - Set

persist_result=True

explicitly - Configure your caching parameters (

cache_key_fn

,

cache_expiration

, etc.) 3. All other tasks will inherit the flow's

persist_result=False

setting This setup gives you the best of both worlds - minimal storage usage for most tasks, but with caching capability where you need it. You can find more details about configuring results and caching in the results documentation and caching documentation.

Ah, I see the issue. Setting

persist_result=False

on the flow alone isn't enough - you need to explicitly set it on each task where you want to disable persistence. Here's the correct way to structure your code:

python
from prefect import flow, task
from datetime import timedelta

# Explicitly disable persistence for non-cached tasks
@task(persist_result=False)
def task_1():
    return "not cached"

@task(persist_result=False)
def task_2():
    return "also not cached"

# Only enable persistence for the cached task
@task(
    persist_result=True,
    cache_key_fn=lambda context: "my-cached-task",
    cache_expiration=timedelta(hours=1)
)
def cached_task():
    return "cached result"

@flow(persist_result=False)
def my_flow():
    task_1()
    task_2()
    cached_task()

If you're still seeing the errors after making these changes, there are two additional things to check: 1. Make sure you've explicitly marked ALL non-cached tasks with

persist_result=False

2. Check if you have any environment variables overriding these settings:

bash
   # Check if this is set to "true" in your environment
   PREFECT_RESULTS_PERSIST_BY_DEFAULT

You can also use Python's context manager to temporarily change persistence settings for a group of tasks:

python
from prefect.context import PrefectObjectRegistry

with PrefectObjectRegistry(
    task_defaults={"persist_result": False}
):
    # All tasks defined in this block will have persist_result=False by default
    @task
    def task_1():
        return "not cached"

This approach might be more manageable if you have many tasks that need the same persistence settings.