Jean Morris
09/15/2025, 2:21 PMMarvin
09/15/2025, 2:21 PMMarvin
09/15/2025, 2:37 PMfrom typing import Literal
from prefect import flow
@flow
def my_flow(env: Literal["dev", "staging", "prod"] = "dev"):
...
- Add numeric bounds with Pydantic annotations (supported by most JSON schema-aware UIs):
from typing import Annotated
from pydantic import Field
from prefect import flow
@flow
def my_flow(
threshold: Annotated[int, Field(ge=0, le=100)] = 50,
):
...
- Or wrap parameters in a Pydantic model if you prefer:
from pydantic import BaseModel, conint
from prefect import flow
class Params(BaseModel):
threshold: conint(ge=0, le=100)
mode: Literal["fast", "full"] = "fast"
@flow
def my_flow(params: Params):
...
- Warn or block at run start (code-level guard)
- Add a guard at the top of your flow to check ranges and either:
- raise a clear error (hard stop), or
- log a warning and continue.
- Example:
from prefect import flow, get_run_logger
@flow
def my_flow(threshold: int):
log = get_run_logger()
if not 0 <= threshold <= 100:
# Choose one:
# Hard stop:
raise ValueError(f"threshold {threshold} is outside 0–100")
# Or soft warn:
# log.warning("threshold %s is outside 0–100; proceeding anyway", threshold)
...
- Require human approval before execution (closest to “Are you sure?”)
- In Prefect Cloud, create an Automation that:
- Triggers on new flow runs (or on scheduled/running state),
- Filters when certain parameter values are out of range,
- Applies a “Require approval” action (and optionally sends a Slack/Email notification).
- The run will pause awaiting approval in the UI; approving continues, rejecting cancels. This gives you a human-in-the-loop confirmation step without modifying the UI form.
If you can share:
- Whether you’re on Prefect Cloud or OSS,
- If you want to block invalid values or just warn/require approval,
- The kinds of constraints you need (enums, numeric ranges, dependent params),
I can give you a tailored snippet and (if on Cloud) click-by-click automation setup.