< Marvin> I try to make cache save both task runs results an Prefect Community #marvin-ai

<@ULVA73B9P> I try to make cache save both task ru...

Ihor Ramskyi

09/26/2025, 3:21 PM

@Marvin I try to make cache save both task runs results and key files to another folder. For task runs results I have to specify @flow(result_storage=block) where block is LocalFileSystem block I already created, containing the path where I want caches to save. Now I got my code running fine and saving cache where I said, but when I try to deploy my flow I get

Copy code

prefect.exceptions.ScriptError: Script at 'main.py' encountered an exception: TypeError('Result storage configuration must be persisted server-side. Please call `.save()` on your block before passing it in.')

The error message is misleading because I already have my block set up (and it gets used fine for running the flow, the problem is with deploying. What could be the fix? My code example:

Copy code

import os
from datetime import timedelta

from prefect import flow, task
from prefect.cache_policies import INPUTS, TASK_SOURCE
from prefect.filesystems import LocalFileSystem

path_block = LocalFileSystem.load("cache-folder")

@task(
    cache_policy=(INPUTS + TASK_SOURCE).configure(key_storage=os.path.join(path_block.basepath, "key_storage")),
    cache_expiration=timedelta(minutes=2)
)
def t():
    return "cache this pls"*1000000


@flow(result_storage=LocalFileSystem.load("cache-folder"))
def main():
    t.submit().result()


if __name__ == "__main__":
    main()

Marvin

09/26/2025, 3:21 PM

✅ thought for 648.0 seconds

Marvin

09/26/2025, 3:32 PM

You’re running into a Prefect 3.x “import-time” check during deploy. What’s happening - When you run locally,

LocalFileSystem.load("cache-folder")

succeeds in your current profile/workspace, so the flow runs and writes results/caches where you expect. - During deploy, Prefect imports your module to discover flows. At import time, it tries to validate

result_storage

. If the deploy process is not connected to the same server/workspace (or cannot resolve the block), Prefect treats the storage as “not persisted” and raises: TypeError: Result storage configuration must be persisted server-side… This is a known pain point with the eager validation in v3; see issue #16641 and a related report issue #16229. Fix options (pick one) 1) Reference the storage block by slug (preferred for deployments) - Don’t call

.load()

at import time. Pass the block slug instead; Prefect will resolve it later.

Copy code

# Use the saved block slug from the UI, e.g. "local-file-system/cache-folder"
@flow(result_storage="local-file-system/cache-folder")
def main():
    t.submit().result()

2) Set the default result storage via settings and drop result_storage from the decorator - Set an environment variable for your deployment/worker:

Copy code

export PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK="local-file-system/cache-folder"

- Then just:

Copy code

@flow
def main():
    t.submit().result()

Docs reference: Settings reference (see PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK) 3) Ensure the deploy process is using the same profile/workspace that contains your block - In the shell where you run

prefect deploy

, verify: -

prefect profile ls

— confirm the active profile -

prefect cloud workspace ls

(Cloud) — confirm the active workspace -

prefect block ls

— confirm the block is visible - If needed, log in and set the workspace before deploying: -

prefect cloud login

and

prefect cloud workspace set <org>/<workspace>

About your cache key storage - Your

cache_policy

path is evaluated at import time too. Since you don’t need a saved block for key storage, avoid calling

.load()

there. Use an absolute path or env var instead:

Copy code

KEY_DIR = os.environ.get("PREFECT_CACHE_KEY_DIR", "/path/to/cache-folder/key_storage")

@task(
    cache_policy=(INPUTS + TASK_SOURCE).configure(key_storage=KEY_DIR),
    cache_expiration=timedelta(minutes=2),
)
def t():
    return "cache this pls" * 1000000

- Note: LocalFileSystem caches are per-machine. If workers run on different machines, they won’t share cache. For shared caching, consider a remote filesystem block instead. Putting it together (safe for deploy)

Copy code

from datetime import timedelta
from prefect import flow, task
from prefect.cache_policies import INPUTS, TASK_SOURCE
import os

KEY_DIR = os.environ.get("PREFECT_CACHE_KEY_DIR", "/absolute/path/to/cache-folder/key_storage")

@task(
    cache_policy=(INPUTS + TASK_SOURCE).configure(key_storage=KEY_DIR),
    cache_expiration=timedelta(minutes=2),
)
def t():
    return "cache this pls" * 1000000

@flow(result_storage="local-file-system/cache-folder")  # saved block slug
def main():
    t.submit().result()

if __name__ == "__main__":
    main()

Docs for reference: - Filesystems: prefect.filesystems - Caching: Caching concepts Quick check: Are you deploying from the same profile/workspace where the “cache-folder” block is saved? If not, switching the profile/workspace or using the block slug/env var above should resolve this.

2 Views

Open in Slack

Previous Next