Great question — here’s the community-backed guidance for Prefect 3.x (assumed) on choosing and using storage blocks like S3Bucket. Short answer - Don’t hardcode credentials or bucket details in code; use saved Blocks and load them at runtime. - It’s common to keep deployments environment-agnostic and select blocks per run via parameters or env/variables. - Passing the block name (string) as a parameter is good practice; passing the Block object itself is not recommended (it won’t serialize cleanly through the API). - For separate input/output buckets, define two blocks (with least-privilege creds) and pass/select them independently. What to tie to a deployment vs. choose at runtime - Good to tie to deployment when: - The environment must be stable/predictable (e.g., a “prod only” deployment). - Least surprise is more important than flexibility. - Good to choose at run-time when: - You want one deployment to work across environments (dev/staging/prod) or across data domains. - You want operators to “pick” via parameters, without redeploying. Recommended patterns 1) Pass block names as flow parameters (most flexible) - Pros: one deployment, easy overrides in UI/API; no rebuild when buckets change. - Cons: you’ll want guardrails/validation so only approved blocks are used.

from prefect import flow
from prefect_aws.s3 import S3Bucket

@flow
def process_data(
    input_block_name: str = "s3-input-dev",
    output_block_name: str = "s3-output-dev",
):
    input_bucket = S3Bucket.load(input_block_name)
    output_bucket = S3Bucket.load(output_block_name)
    # use input_bucket to read and output_bucket to write

2) Select by environment variable or Prefect Variable (clean separation of concerns) - Create blocks like s3-input-dev, s3-input-prod, s3-output-dev, s3-output-prod. - Set ENV in your deployment’s job variables; or store the names in Prefect Variables.

import os
from prefect import flow
from prefect_aws.s3 import S3Bucket
from prefect.variables import Variable

@flow
def process_data():
    env = os.getenv("ENVIRONMENT", "dev")
    input_block_name = Variable.get(f"s3_input_block_{env}") or f"s3-input-{env}"
    output_block_name = Variable.get(f"s3_output_block_{env}") or f"s3-output-{env}"
    input_bucket = S3Bucket.load(input_block_name)
    output_bucket = S3Bucket.load(output_block_name)
    # ...

3) Configure result storage separately - Use result storage for caching/persisting task/flow outputs without hardcoding buckets:

from prefect import flow, task

@task(result_storage="s3-bucket/task-results")  # "block-type-slug/block-name"
def transform(data: bytes) -> bytes:
    # ...
    return b"..."

@flow(result_storage="s3-bucket/flow-results")
def pipeline(input_block_name: str, output_block_name: str):
    # ...
    pass

Notes and best practices - Prefer block names (strings) over Block objects in parameters. Block objects are not JSON-serializable through the API; names are. - Separate input vs output blocks: - Yes, this is common. Define two S3Bucket blocks and keep credentials least-privileged: input may be read-only; output write-only as needed. - This aligns well with security reviews and makes auditing easier. - Keep deployments generic if you expect reuse: - One deployment can serve multiple environments by switching block names or env variables. - If your team wants “fixed” behavior, create per-env deployments (e.g., “pipeline-dev”, “pipeline-prod”) with different default parameters. - For code storage in 3.x: use flow.from_source(...).deploy(...). - Don’t use removed 2.x patterns like Deployment.build_from_flow() or prefect deployment build. Links - AWS S3 block docs: prefect-aws S3 docs - AWS credentials block: AwsCredentials docs

If you can share how you want your users to “choose” (at deployment creation, per run in the UI, or via env), I can suggest a concrete pattern and sample code tailored to your setup.

Short answer: you can’t change a flow’s result_storage from a parameter inside that same flow. In Prefect 3.x, result storage is resolved before the flow run starts (during engine setup), so it isn’t mutable via flow run context or parameters. Supported ways to “choose” result storage - Set it at declaration time on the flow decorator (static). - Use a default from settings (PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK) and vary it per-deployment via env vars. - If you need per-run choice, wrap your logic in a subflow and set the storage in a settings context before calling the subflow. Two practical patterns 1) Per-run choice via wrapper + SettingsContext (recommended)

from prefect import flow
from prefect.settings import Settings
from prefect.settings.context import SettingsContext

@flow  # leave result_storage unset here so it inherits from settings
def inner_flow():
    # your real work
    return "ok"

@flow  # users pass the block name at run time
def driver(result_storage_block: str):  # e.g. "s3-bucket/my-results"
    s = Settings(results={"default_storage_block": result_storage_block})
    with SettingsContext(settings=s):
        return inner_flow()

Alternatively, with temporary_settings and the Setting constant:

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

@flow
def inner_flow():
    return "ok"

@flow
def driver(result_storage_block: str):
    with temporary_settings({PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK: result_storage_block}):
        return inner_flow()

2) Multiple subflows with fixed storage, chosen by param

from prefect import flow

@flow(result_storage="s3-bucket/dev-results")
def inner_dev(): ...

@flow(result_storage="s3-bucket/prod-results")
def inner_prod(): ...

@flow
def driver(env: str = "dev"):
    return inner_prod() if env == "prod" else inner_dev()

Other notes - Passing a Block object as a parameter won’t work; use the block name string (e.g., "s3-bucket/my-results"). - Tasks can have their own result_storage via @task(result_storage="..."), but that also isn’t changeable mid-run by parameters. - If you prefer a per-deployment choice (not per-run), set PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK in the deployment’s job_variables env and keep your flows undecorated for result_storage so they inherit the default. If you tell me whether you need per-run or per-deployment selection, I can tailor a minimal example to your setup.

You’re right to call that out — and I’m sorry for the confusion. I mixed two different things: - Choosing “business data” storage (e.g., which S3 buckets your flow reads/writes): this can be parameterized easily by passing block names and calling

S3Bucket.load(...)

in your code. That’s good practice. - Choosing Prefect’s “result storage” (where the engine persists flow/task results): this is resolved when the run starts and cannot be changed from a parameter inside that same flow run. How to do each cleanly 1) Let users pick input/output S3 blocks (recommended)

from prefect import flow
from prefect_aws.s3 import S3Bucket

@flow
def process_data(input_block: str, output_block: str):
    src = S3Bucket.load(input_block)
    dst = S3Bucket.load(output_block)
    # read from src, write to dst

- This is the pattern I first described — and it’s correct for your data I/O. 2) Let users pick result storage per run (via a wrapper flow) - Because result storage is resolved at run setup, set it before calling your “real” flow:

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

@flow  # leave result_storage unset here so it inherits the default
def inner_flow():
    # your real work
    return "ok"

@flow
def driver(result_storage_block: str):  # e.g. "s3-bucket/my-results"
    with temporary_settings({PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK: result_storage_block}):
        return inner_flow()

- Alternatively, hardcode on the decorator:

@flow(result_storage="s3-bucket/prod-results")
def my_flow(): ...

- Or set a default per-deployment by injecting

PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK

in job env vars. Key takeaway - Passing block names as parameters is great for choosing input/output S3 blocks your code uses. - Result storage must be decided before the flow run begins; use a wrapper or per-deployment settings to vary it. Docs for reference - Results and persistence overview: Persist workflow results - Settings (default result storage): Settings reference - Flow API (result_storage param): Flows API - Result storage utilities: prefect.results API If you share how you want users to choose result storage (per-run in UI, per-deployment, or programmatically), I’ll tailor a minimal example to your setup.