Lester Cayabyab
10/06/2025, 10:53 PMwith_options()Marvin
10/06/2025, 10:53 PMMarvin
10/06/2025, 10:56 PMfrom 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:
# 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:
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.Lester Cayabyab
10/06/2025, 11:04 PMMarvin
10/06/2025, 11:05 PMMarvin
10/06/2025, 11:10 PMfrom typing import Literal
from prefect import flow
@flow
def ingest(env: Literal["dev", "staging", "prod"] = "dev"):
print(f"Running in {env}")
- Using an Enum
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:
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.Lester Cayabyab
10/06/2025, 11:14 PMMarvin
10/06/2025, 11:14 PMMarvin
10/06/2025, 11:15 PMfrom 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)
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.Lester Cayabyab
10/07/2025, 2:55 PMMarvin
10/07/2025, 2:55 PMMarvin
10/07/2025, 2:57 PMimport 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:
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.