<@ULVA73B9P> is it possible to pass a option in fl...
# ask-marvin
l
@Marvin is it possible to pass a option in flow where there is a list of options but there is already a default value on it using
with_options()
m
thought for 196.1 seconds
Short answer: with_options() does not merge list-type options — it replaces them. - For example, if a Task already has default tags, passing tags=[...] in with_options() will overwrite the existing tags. Same goes for other list options (e.g., task/flow state hooks). To extend instead of overwrite, combine the existing value with your additions before calling with_options():
Copy code
from prefect import task

@task(tags=["default"])
def my_task():
    ...

# Overwrites tags -> becomes ["extra"]
my_task_v2 = my_task.with_options(tags=["extra"])

# Extends tags -> becomes ["default", "extra"]
existing = list(my_task.tags or [])
my_task_v3 = my_task.with_options(tags=existing + ["extra"])
For flows, the same pattern applies to list options like on_failure/on_completion hooks:
Copy code
# Example idea (combine existing hooks with a new one)
flow_v2 = flow_v1.with_options(on_failure=[*(flow_v1.on_failure or []), my_hook])
Alternative for tags at run time: use the tags context to merge tags temporarily without changing the task/flow definition:
Copy code
from prefect import tags

with tags("extra", "team-x"):
    my_flow()  # runs created in this context include these tags in addition to existing ones
Docs on tags/context: Prefect 3.x context and tags If you can share which specific option you’re trying to extend (e.g., tags, on_failure hooks) and whether it’s on a Task or Flow, I can tailor an exact snippet.
l
@Marvin For example I want to have a dropdown in the custom run variables how to do that?
m
thought for 328.8 seconds
Yes — if you’re on Prefect 3.x, you can get a dropdown in the “Run” form by constraining your flow parameter’s type to a fixed set of choices. The UI will render it as a select and pre-select your default. Two easy ways: - Using typing.Literal
Copy code
from typing import Literal
from prefect import flow

@flow
def ingest(env: Literal["dev", "staging", "prod"] = "dev"):
    print(f"Running in {env}")
- Using an Enum
Copy code
from enum import Enum
from prefect import flow

class Env(str, Enum):
    dev = "dev"
    staging = "staging"
    prod = "prod"

@flow
def ingest(env: Env = Env.dev):
    print(f"Running in {env}")
When you click “Run” in the UI,
env
will appear as a dropdown with your allowed values, and the default will be selected automatically. If instead you want a dropdown mid-run (i.e., pause the flow and ask a user to choose), you can use an Enum with
pause_flow_run
to prompt for input:
Copy code
from enum import Enum
from prefect import flow
from prefect.flow_runs import pause_flow_run

class Decision(str, Enum):
    proceed = "proceed"
    abort = "abort"

@flow
def pipeline():
    decision = pause_flow_run(wait_for_input=Decision, key="Approval")
    if decision == Decision.proceed:
        ...
More details: - Form building and choices: Form building - Interactive prompts: Interactive workflows Is your dropdown for parameters in the Run dialog, or do you want to pause and ask during the run? I can tailor the snippet to your exact use case.
l
Can you make the enum dynamic value that I pass? @Marvin
m
thought for 37.2 seconds
Great question. Short answer: it depends on where you want the dropdown. - Run dialog (before the flow starts): the choices must be known when the deployment is created. You can generate an Enum from a list, but it won’t be “live” per run — you’d need to redeploy to update the options. - During the run (interactive prompt): yes, you can make the choices truly dynamic by computing them at runtime and pausing for input with a dynamically built Enum. Two patterns: 1) Static at run dialog (choices fixed at deploy time)
Copy code
from enum import Enum
from prefect import flow

def build_enum(name, options: list[str]):
    return Enum(name, {o: o for o in options})

# Compute at import/deploy time
OPTIONS = ["dev", "staging", "prod"]  # e.g., loaded from a file/DB/env
Env = build_enum("Env", OPTIONS)

@flow
def ingest(env: Env = next(iter(Env))):  # default is first option
    print(env.value)
Note: to change the dropdown options in the Run dialog, update OPTIONS and redeploy so the UI gets the new schema. 2) Truly dynamic during the run (interactive)
Copy code
from enum import Enum
from prefect import flow
from prefect.flow_runs import pause_flow_run

def make_enum(name, options: list[str]):
    return Enum(name, {o: o for o in options})

@flow
def pipeline():
    # Compute at runtime
    options = fetch_available_targets()  # e.g., hit an API/DB
    Target = make_enum("Target", options)

    target = pause_flow_run(wait_for_input=Target, key="Pick a target")
    print(f"Selected: {target.value}")
Here, the UI will render a dropdown based on the Enum you just created at runtime. References: - Interactive workflows and prompts: Interactive workflows - Form building (how types drive UI fields): Form building Which experience do you want: a dropdown in the Run dialog (pre-run) or a prompt during the run? If you share how you’re deriving the choices, I can sketch the exact code.
l
@Marvin can you dynamically set the list of options on flow parameter and also set the default value
m
thought for 118.5 seconds
Yes, with one caveat. - Run dialog (flow parameters): you can generate the choices and default dynamically at import/deploy time, but the UI won’t update per-run. If the list changes, you’ll need to redeploy so the new schema reaches the UI. - True per-run dynamic choices: use an interactive prompt during the run (pause for input), not a parameter. Pattern for parameters (dynamic at deploy time) — build a str Enum from a list and set a default:
Copy code
import re
from enum import Enum
from prefect import flow

def fetch_options() -> list[str]:
    # e.g. pull from a DB, file, env var, etc.
    return ["dev", "staging", "prod"]

def make_choice_enum(name: str, options: list[str]) -> type[Enum]:
    # Ensure valid Enum member names; UI will display the string values
    def safe(n: str) -> str:
        return re.sub(r"\W|^(?=\d)", "_", n).upper()
    members = {safe(o): o for o in options}
    return Enum(name, members, type=str)  # creates a str-backed Enum

OPTIONS = fetch_options()
Env = make_choice_enum("Env", OPTIONS)

DEFAULT = "dev" if "dev" in OPTIONS else OPTIONS[0]

@flow
def ingest(env: Env = Env(DEFAULT)):
    print(f"Selected: {env.value}")
- The Run dialog will render
env
as a dropdown with the values in
OPTIONS
. - The default shown will be whatever you set in the function signature (here
DEFAULT
). - To change the list or default shown in the Run dialog later, update the options/default and redeploy. If you really need the list to be computed per-run without redeploy, switch to an interactive prompt mid-flow:
Copy code
from enum import Enum
from prefect import flow
from prefect.flow_runs import pause_flow_run

def make_choice_enum(name, options):
    import re
    return Enum(
        name,
        {re.sub(r"\W|^(?=\d)", "_", o).upper(): o for o in options},
        type=str,
    )

@flow
def pipeline():
    options = fetch_options()  # computed at runtime
    Choice = make_choice_enum("Choice", options)
    selection = pause_flow_run(wait_for_input=Choice, key="Pick an option")
    print(f"User picked: {selection.value}")
Docs: - Form building (how parameter types become UI fields): Form building - Interactive prompts during runs: Interactive workflows Do you need the options to change per run without redeploy, or is “update list and redeploy” acceptable? I can tailor the approach based on that.