Great question. In Prefect 3.x you can keep a single code path and switch persistence between local and GCS with settings, not conditionals. Here’s a simple, low-overhead pattern that works for both development and CI. High-level approach - Save two blocks once: a LocalFileSystem block for local dev/tests and a GcsBucket block for cloud. - In code, set

persist_result=True

on your flow/tasks but do NOT hardcode

result_storage

. Let the environment pick the default storage block via

PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK

. - In tests, parametrize storage by temporarily setting that environment-backed setting to the local or GCS block slug. Your flow code stays unchanged. 1) One-time block setup Run these once (locally or in a setup script). For GCS you’ll need

prefect-gcp

and credentials (e.g.,

GOOGLE_APPLICATION_CREDENTIALS

set to a service account key).

# Local block
from prefect.filesystems import LocalFileSystem
LocalFileSystem(basepath="/tmp/prefect-results").save("local-results", _sync=True)

# GCS block (pip install 'prefect-gcp' first)
from prefect_gcp.cloud_storage import GcsBucket
GcsBucket(bucket="my-bucket", bucket_folder="results/").save("gcs-results", _sync=True)

Verify blocks exist:

prefect block ls

2) Keep the flow code simple Just enable persistence. Let the default storage block be provided by settings/env.

from prefect import flow, task

@task
def compute(x: int) -> int:
    return x * 2

@flow(persist_result=True)  # no result_storage hardcoding
def my_flow(n: int = 3):
    return [compute(i) for i in range(n)]

if __name__ == "__main__":
    # Local: export PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK="local-file-system/local-results"
    # GCS:   export PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK="gcs-bucket/gcs-results"
    my_flow()

Optional defaults via env: - Use local by default:

export PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK="local-file-system/local-results"

- Use GCS by default:

export PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK="gcs-bucket/gcs-results"

- If you want persistence on without decorating every flow:

export PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

3) Tests that cover both backends (no overengineering) Parametrize tests and switch the default storage block with

temporary_settings

. No if/else in your flow code.

import pytest
from prefect.settings import temporary_settings, PREFECT_DEFAULT_RESULT_STORAGE_BLOCK
from my_package.flows import my_flow  # wherever your flow lives

@pytest.mark.parametrize("block_slug", [
    "local-file-system/local-results",
    "gcs-bucket/gcs-results",
])
def test_persistence_across_backends(block_slug):
    with temporary_settings({PREFECT_DEFAULT_RESULT_STORAGE_BLOCK: block_slug}):
        out = my_flow(3)
        assert out == [0, 2, 4]

Tips and gotchas - Make sure blocks are saved before referencing them by slug. - For GCS tests, ensure credentials are present; otherwise mark/skip:

pytest.importorskip("prefect_gcp")

or skip the GCS param if

GOOGLE_APPLICATION_CREDENTIALS

isn’t set. - If you prefer absolutely no writes during unit tests, disable persistence:

with temporary_settings({"PREFECT_RESULTS_PERSIST_BY_DEFAULT": False}):
      ...

- You can still override storage per-task/flow when needed:

@task(result_storage="gcs-bucket/gcs-results", persist_result=True)

Why this stays simple - Only one code path; environment chooses storage. - Zero branching in flow code. - Tests toggle a single setting to cover both local and GCS. Helpful docs - Results (Prefect 3.x) - List and inspect blocks via CLI: -

prefect block ls

-

prefect block inspect <slug>

If you want, I can share a quick script to create both blocks and print the correct slugs for your repo/CI.

Short answer - You set the object “path” with the task’s

result_storage_key

. - If you don’t set it, Prefect generates a unique UUID key per run — this is concurrency-safe and avoids overwrites. - It doesn’t matter “where” they live for correctness; organization is up to you. The only time it matters is if you hand-pick keys that could collide under concurrency. What to use in practice - If you just want it to work under concurrency: don’t set

result_storage_key

(or include a unique value like

{task_run.id}

). - If you need deterministic, predictable paths: set

result_storage_key

with a template, but ensure uniqueness under concurrent runs. Key details -

result_storage_key

exists on tasks (not flows) in Prefect 3.x. - Key templating supports runtime context: -

{task_run.id}

,

{task_run.name}

,

{flow_run.id}

,

{flow_run.flow_name}

,

{parameters[...]}

- The key is a relative path under the configured storage: - LocalFileSystem:

<basepath>/<result_storage_key>

- GCS:

gs://<bucket>/<bucket_folder>/<result_storage_key>

- You can include slashes to create folders. Concurrency guidance - Safe default (no collisions):

@task(persist_result=True)  # no result_storage_key => unique UUID per run
  def compute(x): ...

- Predictable but still safe:

@task(persist_result=True, result_storage_key="flows/{flow_run.flow_name}/{flow_run.id}/tasks/{task_run.name}-{task_run.id}")
  def compute(x): ...

- Deterministic (idempotent) — use with care for concurrency (can overwrite if two runs hit the same key at the same time):

@task(
      persist_result=True,
      result_storage_key="dedup/{task_run.task_name}/user={parameters[user_id]}"
  )
  def build_profile(user_id: str): ...

If you expect concurrent runs with the same

user_id

, add a unique suffix (e.g.,

{flow_run.id}

) or serialize access externally (e.g., limit concurrency) to avoid last-write-wins. Does it matter where they live? - For Prefect: no. Prefect will read/write via the storage and key you provide. - For you: use a simple hierarchy so you can find things (e.g.,

flows/{flow_run.flow_name}/{flow_run.id}/tasks/...

) and avoid accidental collisions. Minimal examples - Local/GCS with safe keys:

from prefect import task

  @task(persist_result=True, result_storage_key="flows/{flow_run.flow_name}/{flow_run.id}/tasks/{task_run.name}-{task_run.id}")
  def compute(x: int) -> int:
      return x * 2

- Deterministic, id-based (only if you control concurrency):

@task(persist_result=True, result_storage_key="profiles/{parameters[user_id]}")
  def build_profile(user_id: str): ...

Docs: Results (Prefect 3.x) If you share a concrete pattern you want (e.g., “group by flow then task and keep deterministic per input”), I can suggest an exact

result_storage_key

template that’s both organized and concurrency-safe.

It looks like your override isn’t being applied, so the run is falling back to a different slug (

gcs-bucket/gcs-results

) and failing to find that block. A few key points for Prefect 3.x: - In deployments, environment variables must be set under work_pool.job_variables.env (not a top-level environment key). - The setting to pick the default result storage is

PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK

(alias:

PREFECT_DEFAULT_RESULT_STORAGE_BLOCK

). - The block slug must exactly match

<block-type-slug>/<block-name>

and the block must exist. What to change 1) Put your setting under job_variables.env in your deployment config:

work_pool:
  name: your-pool
  job_variables:
    env:
      PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK: "gcs-bucket/ingestion-service-results"
      PREFECT_RESULTS_PERSIST_BY_DEFAULT: "true"   # optional: enable persistence globally

2) Make sure the block exists and the name matches exactly:

prefect block ls
# confirm there is a 'gcs-bucket' block named 'ingestion-service-results'

If not, create/save it:

# ensure GCP blocks are registered
prefect block register -m prefect_gcp.cloud_storage

# in Python
from prefect_gcp.cloud_storage import GcsBucket
GcsBucket(bucket="your-bucket", bucket_folder="results/").save("ingestion-service-results", _sync=True)

3) Alternative (explicit per-flow): set result_storage on your flow/tasks so you’re not relying on env for this:

from prefect import flow

@flow(persist_result=True, result_storage="gcs-bucket/ingestion-service-results")
def my_flow(...):
    ...

Why you saw the 404 - The worker tried to resolve

gcs-bucket/gcs-results

, which means the setting in effect for that run was pointing to that slug. Common causes: - The deployment used

environment:

(ignored), not

job_variables.env:

. - A different environment variable (on the worker host) is overriding your intended value. - The specified block name doesn’t exist. Quick verification inside a run Add a tiny check to confirm the effective value:

from prefect import flow
from prefect.settings import get_current_settings, PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK

@flow
def debug_settings():
    s = get_current_settings()
    print("RESULTS BLOCK:", s.get(PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK))

debug_settings()

If you share your deployment YAML snippet (work_pool → job_variables → env section) and confirm the exact block name you saved, I can spot-check it. Also, please run:

prefect block ls

and paste the line for your GCS block. Docs for reference: - Results (Prefect 3.x)