Great question. Since you’re on Prefect 2.x, the simplest way to use a LocalFileSystem only when running locally is to choose the result storage at run time based on environment, not on Prefect runtime state (because decorator options are evaluated at import time). Below are two clean patterns you can use. Pattern A: Environment flag to enable local results only when running locally - Set an env var locally (e.g., LOCAL_RUN=1) and leave it unset in Cloud. - Persist results to a LocalFileSystem when local; don’t persist on Cloud.

import os
from prefect import flow, task
from prefect.filesystems import LocalFileSystem
from prefect.serializers import JSONSerializer

LOCAL_RUN = os.getenv("LOCAL_RUN", "0") == "1"

# Only persist + write to local disk when running locally
RESULT_STORAGE = LocalFileSystem(basepath="./.prefect/results") if LOCAL_RUN else None
PERSIST = LOCAL_RUN

@task
def do_work():
    return {"ok": True}

@flow(
    persist_result=PERSIST,
    result_storage=RESULT_STORAGE,
    result_serializer=JSONSerializer()
)
def my_flow():
    return do_work()

if __name__ == "__main__":
    # Local run: enable local results
    # LOCAL_RUN=1 python my_flow.py
    my_flow()

Notes: - When LOCAL_RUN is unset (Cloud deployment), this flow won’t persist results. - If your flow/tasks rely on features that require persistence (e.g., retries, certain caching configs), Prefect may auto-require persistence. In that case, keep this pattern and set a remote filesystem block for deployments (e.g., a GCS-backed filesystem) instead of disabling persistence entirely. Pattern B: Use with_options for local runs only - Keep the “base” flow configured for Cloud (no persistence). - Wrap it with with_options only when running locally.

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

@task
def do_work():
    return {"ok": True}

@flow(persist_result=False)  # Default for Cloud runs
def my_flow():
    return do_work()

# Local variant that persists to LocalFileSystem
my_flow_local = my_flow.with_options(
    persist_result=True,
    result_storage=LocalFileSystem(basepath="./.prefect/results")
)

if __name__ == "__main__":
    # Local run: use the local variant
    my_flow_local()

How to detect “local” vs “deployment”? - At decoration time (when the module is imported), Prefect’s runtime context isn’t available, so don’t try to branch on prefect.runtime there. - If you need to branch inside the flow at execution time (for logging, etc.), you can check:

from prefect.runtime import deployment
  IS_DEPLOYMENT = deployment.id is not None

But use the environment variable or with_options patterns above for configuring result storage. Follow-up questions for you: - On Cloud runs, do you want to fully disable persistence, or would you prefer persisting to GCS? If you want GCS, I can share a minimal result_storage setup that uses a GCS-backed filesystem for deployments while keeping LocalFileSystem for local runs.

Got it! If your goal is to “dump data across the flow” to local disk using Prefect 2.x, you’ve got two complementary tools: 1) Use LocalFileSystem directly inside tasks to write files anywhere under a base directory - Best for ad-hoc artifacts or side outputs you want to control precisely. 2) Use Prefect’s result persistence with LocalFileSystem so task returns are automatically saved - Best when you want Prefect to handle writing task results for caching/retries/checkpointing. Below are minimal, correct examples for both. A) Write files manually with LocalFileSystem inside tasks - LocalFileSystem supports: - write_path(path, bytes) - read_path(path) -> bytes - put_directory(local_path, to_path) - get_directory(from_path, local_path)

from prefect import flow, task
from prefect.filesystems import LocalFileSystem
import json

ARTIFACTS = LocalFileSystem(basepath="./.prefect/artifacts")

@task
def write_artifacts(run_name: str, data: dict) -> str:
    # Convert to bytes and write
    content = json.dumps(data, indent=2).encode()
    path = f"runs/{run_name}/data.json"
    saved_path = ARTIFACTS.write_path(path, content)
    return saved_path

@task
def read_artifacts(run_name: str) -> dict:
    content = ARTIFACTS.read_path(f"runs/{run_name}/data.json")
    return json.loads(content.decode())

@flow
def my_flow():
    run_name = "example-run"
    p = write_artifacts(run_name, {"status": "ok", "n": 3})
    loaded = read_artifacts(run_name)
    return {"saved": p, "loaded": loaded}

if __name__ == "__main__":
    my_flow()

B) Let Prefect persist task results to LocalFileSystem - Persist task returns automatically - Use result_storage_key to control the file names/paths per task run

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

RESULTS = LocalFileSystem(basepath="./.prefect/results")

@task(
    persist_result=True,
    result_storage=RESULTS,
    result_storage_key="runs/{flow_run.name}/step1.json"
)
def step1(x: int) -> dict:
    return {"step": 1, "x": x}

@task(
    persist_result=True,
    result_storage=RESULTS,
    result_storage_key="runs/{flow_run.name}/step2_{parameters[y]}.json"
)
def step2(y: str, prev: dict) -> dict:
    return {"step": 2, "y": y, "prev": prev}

@flow(
    persist_result=True,                      # also persist the flow return
    result_storage=RESULTS,
    result_serializer=JSONSerializer(),
)
def pipeline(x: int = 1, y: str = "A"):
    a = step1(x)
    b = step2(y, a)
    return {"final": b}

if __name__ == "__main__":
    pipeline()

Tips and gotchas - Local vs Cloud: If you only want LocalFileSystem for local runs, set an env var like LOCAL_RUN=1 locally and branch your configuration at import time:

import os
  USE_LOCAL = os.getenv("LOCAL_RUN") == "1"
  RESULTS = LocalFileSystem(basepath="./.prefect/results") if USE_LOCAL else None

  @flow(persist_result=USE_LOCAL, result_storage=RESULTS)
  def my_flow(): ...

- Path structure: Organize by run using

{flow_run.name}

or parameters in

result_storage_key

. - Data types: When writing manually with

write_path

, encode to bytes. With result persistence, Prefect handles serialization via

result_serializer

(e.g., JSONSerializer). - Directory ops: Use put_directory/get_directory for bulk copies of folders. If you share how you want the files organized (by run, by date, by task), I can tailor the

result_storage_key

patterns or write_path paths for your exact layout.