Great question. In Prefect 3.x, the easiest way to make “full refresh” toggleable from the UI is to expose it as a boolean flow parameter, then pass that flag through to PrefectDbtRunner when you build the dbt command. Here’s a minimal pattern you can copy:

from typing import Annotated, Optional
from pydantic import Field
from prefect import flow, task
from prefect_dbt.core.runner import PrefectDbtRunner

@task
def run_dbt(full_refresh: bool = False, selector: Optional[str] = None, command: str = "run"):
    runner = PrefectDbtRunner()

    # Build the dbt CLI args list
    args = [command]  # e.g., "run" or "build"
    if selector:
        args += ["--select", selector]
    if full_refresh:
        args += ["--full-refresh"]

    # This calls dbt under the hood and streams results/logs back to Prefect
    return runner.invoke(args)

@flow
def dbt_flow(
    full_refresh: Annotated[bool, Field(description="Force rebuild of incremental models")] = False,
    selector: Optional[str] = None,
    command: Annotated[str, Field(description="dbt subcommand to run (e.g., run or build)")] = "run",
):
    return run_dbt(full_refresh=full_refresh, selector=selector, command=command)

if __name__ == "__main__":
    dbt_flow()

Why this works - Boolean flow parameters show up as a checkbox in the UI. When you click “Custom run” on the deployment, you can toggle

full_refresh

on/off. -

PrefectDbtRunner.invoke([...])

accepts a list of dbt CLI args, so

--full-refresh

can be added conditionally. We verified the signature for

invoke(args: list[str], **kwargs)

. Deploy with a default and allow UI overrides

from prefect import flow

# in a separate deploy.py or under if __name__ == "__main__":
flow.from_source(
    source="<https://github.com/your-org/your-repo.git>",
    entrypoint="flows/dbt_flow.py:dbt_flow",
).deploy(
    name="prod",
    work_pool_name="your-pool",
    parameters={"full_refresh": False},  # default for scheduled/manual runs
    tags=["dbt"],
)

How to trigger a full refresh from the UI - Open your deployment in the UI - Click “Run” → “Custom run” - Toggle the “full_refresh” checkbox to true - Submit Optional: CLI override for ad-hoc runs - You can also trigger a full refresh from the CLI by overriding parameters:

prefect deployment run "dbt_flow/prod" --param full_refresh=true --watch

Note: values are interpreted as JSON, so use

true/false

rather than

True/False

. Notes and links - Prefect dbt integration: Prefect dbt docs - Building parameter forms (descriptions, enums, etc.): Form building - Python-based deploys: Deploy via Python - CLI reference for deployments: Deployment CLI - Blog: dbt + Prefect in minutes Would you like this adapted for

dbt build

(instead of

run

) or to include selectors/env-specific flags by default?

Totally doable. Since you’re calling

PrefectDbtRunner

from a task that multiple flows use, the cleanest approach is: - Add a

full_refresh: bool = False

parameter to your task - Have each flow expose a

full_refresh

parameter (checkbox in the UI) and pass it to the task - In the task, append

--full-refresh

only for `run`/`build` commands (and skip for

retry

) Here’s how you can adapt your code. Task: add a

full_refresh

flag and inject it into the dbt args

import os
import shlex
from typing import Optional

from prefect import task, get_run_logger
from prefect_dbt.core.runner import PrefectDbtRunner
from prefect_snowflake.credentials import SnowflakeCredentials

@task(name="Execute dbt")
def execute_dbt(
    commands: list[str] | None = None,
    full_refresh: bool = False,
) -> None:
    """
    Executes dbt commands using Prefect's PrefectDbtRunner.

    Args:
        commands (list[str] | None): List of dbt commands to execute (e.g., ["build -s +model"])
        full_refresh (bool): If True, pass --full-refresh to relevant dbt commands.
    """

    eu_credentials = SnowflakeCredentials.load("wiwnhis-eu-service-credentials")

    # initialize env vars
    os.environ["DBT_PROFILES_DIR"] = "dbt_models"
    os.environ["DBT_PROJECT_DIR"] = "dbt_models"
    os.environ["SNOWFLAKE_ACCOUNT"] = eu_credentials.account
    os.environ["SNOWFLAKE_PRIVATE_KEY"] = (
        eu_credentials.private_key.get_secret_value().decode("utf-8")
    )

    logger = get_run_logger()
    <http://logger.info|logger.info>("Building dbt models...")

    # Always prepend 'deps' to the commands list
    if commands is None:
        commands = ["build"]
    commands = ["deps"] + commands

    runner = PrefectDbtRunner(include_compiled_code=True)

    # Helper: add --full-refresh to eligible commands only
    def with_full_refresh_if_applicable(args: list[str]) -> list[str]:
        if not full_refresh:
            return args
        if not args:
            return args
        cmd = args[0]
        # Apply to run/build; skip 'deps', 'retry', etc.
        if cmd in {"run", "build"} and "--full-refresh" not in args:
            return args + ["--full-refresh"]
        return args

    # First runs: execute each command; reruns: 'retry' failed nodes
    # (Preserve your existing behavior; if you have custom get_run_count, keep it.)
    try:
        from prefect.runtime import flow_run
        run_count = flow_run.run_count  # available in Prefect 3.x
    except Exception:
        # Fallback if you're using your own get_run_count() helper
        run_count = 1

    if run_count == 1:
        for command in commands:
            args = shlex.split(command)
            args = with_full_refresh_if_applicable(args)
            runner.invoke(args)
    else:
        # For retries, do not pass --full-refresh
        runner.invoke(["retry"])

Flow: expose

full_refresh

and pass it to the task

from prefect import flow
from utils.utils_snowflake import execute_dbt

@flow(
    retries=1,
    retry_delay_seconds=3,
    timeout_seconds=3600,
    name="EU Newsletter Clicks(with Inflation Flags)",
)
def eu_clicks_flow(full_refresh: bool = False):
    # Your selector stays embedded in the command string
    execute_dbt(["build -s +clicks_by_user_agent"], full_refresh=full_refresh)

if __name__ == "__main__":
    eu_clicks_flow()

How to trigger full refresh from the UI - Deploy this flow (so Prefect can render parameters in the UI) - Open the deployment → Run → Custom run - Toggle the “full_refresh” checkbox to true → Submit Optional: CLI override for ad-hoc runs

prefect deployment run "EU Newsletter Clicks(with Inflation Flags)/<your-deployment-name>" --param full_refresh=true --watch

Notes and docs - Prefect dbt integration: Prefect dbt docs - Parameter forms in the UI: Form building

- Deploying via Python: Deploy via Python - Deployment CLI: Deployment CLI If you’d prefer to avoid string parsing with

shlex

, you could refactor to pass structured command pieces (e.g., command="build", select="+model")—happy to sketch that out too.

Thanks for the details — the error tells us exactly what’s happening: - Your flow function currently takes no parameters (Function expects parameters []), but your Custom Run is sending one:

dbt_full_refresh

. - With the “Enforce parameter schema” flag enabled, Prefect validates the parameters passed to the deployment against the flow function’s signature and raises

SignatureMismatchError

when they don’t match. How to fix it (recommended): 1) Add the parameter to your flow function Update your flow to accept

dbt_full_refresh

(and give it a default if you want):

from prefect import flow

@flow
def fields_of_interest_flow(dbt_full_refresh: bool = False):
    if dbt_full_refresh:
        # do a full refresh path
        ...
    else:
        # normal path
        ...

2) Redeploy and (optionally) set a default in your deployment Since you’re using prefect.yaml, add a default under

parameters

so the UI picks it up:

deployments:
  - name: fields_of_interest
    entrypoint: flows/fields_of_interest.py:fields_of_interest_flow
    description: "Retrieve all users and their fields of interests (pivoted by column)."
    parameters:
      dbt_full_refresh: false
    work_pool:
      name: default
      work_queue_name: default
      job_variables: {}
    tags:
      - prod
      - eu
      - subscribers
      - interests
      - owner:U04Q4QE5ENN
    schedules:
      - cron: 0 9 * * 1
        timezone: UTC
        day_or: true
        active: true

Then run

prefect deploy

from your project root to update the deployment so the UI knows about the new parameter. Alternative options (not preferred long-term): - Remove the

dbt_full_refresh

parameter from the Custom Run form if you don’t actually need it. - Turn off “Enforce parameter schema” for that run. This removes UI-time validation, but if your flow still doesn’t accept that parameter, it won’t be usable by the flow. The correct fix is to add the parameter to the flow signature. Helpful links: - Deployments with prefect.yaml: prefect.yaml guide - Flow/deployment parameter handling (API): Flow API reference - CLI deploy reference: prefect deploy CLI If you can share your current

fields_of_interest_flow

signature, I can suggest the exact change.