How to Configure Result Storage and Persistence in Prefect Flows

Prefect configures result storage through the ResultsSettings model and ResultStore class, allowing you to specify serializers, storage backends, and persistence behavior globally, per-flow, or per-task.

Configuring result storage and persistence in Prefect flows ensures that task outputs are reliably serialized and stored for downstream consumption, caching, or debugging. The PrefectHQ/prefect repository provides a flexible hierarchy of settings that govern how results are written, where they are stored, and whether persistence is enabled by default.

Understanding the ResultsSettings Configuration Model

The foundation of result configuration lies in the ResultsSettings model, defined in src/prefect/settings/models/results.py. This model exposes four critical options that control default behavior across your Prefect instance:

• default_serializer – Specifies the serialization method (e.g., "pickle") when a task or flow does not declare one explicitly.
• persist_by_default – A global boolean flag that enables result persistence when no explicit setting is provided. Defaults to False.
• default_storage_block – The block slug (block-type/block-name) pointing to the storage block used as the default result store. Defaults to None.
• local_storage_path – A fallback local directory (defaulting to $PREFECT_HOME/storage) used when no remote block is configured.

These settings are accessed at runtime via prefect.settings.context.get_current_settings() and influence resolution logic throughout the execution lifecycle.

How Prefect Resolves Default Result Storage at Runtime

When a flow executes, Prefect determines the appropriate storage backend through a hierarchical resolution process implemented in _aget_default_result_storage within src/prefect/results.py. The system evaluates candidates in the following priority order:

1. User-provided block slug – The value specified in results.default_storage_block settings.
2. Server-configured default block – Retrieved from the Prefect server via read_server_default_result_storage.
3. Local filesystem fallback – A LocalFileSystem block rooted at results.local_storage_path.

The resolved storage is cached in _default_storages to prevent redundant block lookups during flow execution.

Controlling When Results Persist

The function should_persist_result() (lines 91–98 in src/prefect/results.py) implements the logic for determining whether a specific result should be persisted:

• If an active run context exists (TaskRunContext or FlowRunContext), the function respects the context-specific persist_result flag.
• In the absence of a context, it falls back to the global results.persist_by_default setting.
• When a default storage block is configured (either via settings or server configuration), persistence is implicitly enabled regardless of other flags.

The ResultStore Runtime API

ResultStore, beginning at line 37 in src/prefect/results.py, serves as the concrete interface for reading and writing results during task and flow execution. This class encapsulates:

• result_storage – The storage block (e.g., S3, Azure Blob, or local filesystem) where serialized results are written.
• metadata_storage – An optional separate block for metadata when users require decoupled storage layouts.
• serializer – The Serializer instance selected via get_default_result_serializer() (lines 63–68).
• lock_manager – An optional component providing transactional isolation for concurrent access.

ResultStore exposes high-level asynchronous and synchronous methods including write(), read(), awrite(), and aread(), along with lock-related APIs for advanced use cases.

Managing Defaults with the Prefect CLI

Prefect provides dedicated CLI commands in src/prefect/cli/result_storage.py for inspecting and modifying the server-wide default result storage configuration:

# Display the current server-wide default storage block

prefect result-storage inspect

# Set a new default storage block for all flows

prefect result-storage set <BLOCK_SLUG>

# Remove the server-wide configuration

prefect result-storage clear

These commands interact directly with the read_server_default_result_storage and update_server_default_result_storage endpoints used by the runtime resolution logic.

Practical Configuration Examples

Setting a Global Default Storage Block in Code

You can programmatically override result storage settings for the entire process duration:

from prefect import flow
from prefect.settings import PrefectSettings

# Configure global defaults for this process

settings = PrefectSettings()
settings.results.default_storage_block = "s3/my-prefect-results"
settings.results.persist_by_default = True

This configuration forces all subsequent flows to use the specified S3 block unless explicitly overridden at the flow or task level.

Configuring Storage and Serializers Per-Flow

Override storage behavior for specific tasks or flows using decorators:

from prefect import flow, task
from prefect.filesystems import LocalFileSystem
from prefect.serializers import JsonSerializer

@task(result_storage=LocalFileSystem(basepath="/tmp/prefect-results"),
      result_serializer=JsonSerializer())
def compute(x: int) -> int:
    return x * 2

@flow(result_storage="gcs/my-gcs-results")
def my_flow():
    a = compute(3)
    b = compute(5)
    return a + b

if __name__ == "__main__":
    my_flow()

In this example, compute writes results to a local directory using JSON serialization, while the flow uses a Google Cloud Storage block for any results without task-level storage defined.

Setting Server-Wide Defaults via CLI

Configure a shared backend for all deployments in your Prefect server:

prefect result-storage set s3/prefect-default-results

Once set, any flow that does not specify its own result_storage will automatically use this S3 block.

Accessing ResultStore Directly for Custom Logic

For advanced scenarios requiring custom key handling, access the ResultStore directly:

from prefect import get_run_context, task
from prefect.results import get_result_store

@task
def my_task():
    store = get_result_store()
    store.write({"value": 42})
    record = store.read("some-key")
    return record.result

This pattern allows direct manipulation of the storage layer while respecting the configured serializers and backends.

Summary

• ResultsSettings in src/prefect/settings/models/results.py defines global defaults for serializers, persistence behavior, and storage blocks.
• Prefect resolves storage through a three-tier hierarchy: user settings, server configuration, and local filesystem fallback.
• The should_persist_result() function determines persistence based on context flags and global settings.
• ResultStore in src/prefect/results.py provides the runtime API for reading and writing results with support for separate metadata storage and locking.
• The prefect result-storage CLI commands manage server-wide default storage configurations.
• Configuration can be applied globally via settings, per-flow via decorators, or server-wide via CLI commands.

Frequently Asked Questions

What happens if I don't configure any result storage?

According to the resolution logic in src/prefect/results.py, Prefect falls back to a LocalFileSystem block rooted at results.local_storage_path, which defaults to $PREFECT_HOME/storage. Results are only persisted if persist_by_default is set to True or if explicitly enabled on a specific task or flow.

How does Prefect decide whether to persist a result?

The should_persist_result() function checks for an active run context first, respecting any persist_result flag set there. If no context exists, it uses the global results.persist_by_default setting. Additionally, configuring a default storage block implicitly enables persistence for that run.

Can I use different storage backends for different tasks in the same flow?

Yes. You can specify different result_storage blocks on individual tasks using the decorator parameter, while setting a different default at the flow level. Each task will use its configured storage, falling back to the flow's storage only when no task-level storage is specified.

Where does Prefect store the configuration for default result storage?

Default storage configuration resides in two places: the local ResultsSettings model accessed via prefect.settings, and the Prefect server database for deployment-wide defaults. The CLI commands in src/prefect/cli/result_storage.py modify the server-side configuration, while runtime code can override settings programmatically.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}