@Marvin the endpoint patch /api/deployments/{id} returns {‘detail’: ‘Method Not Allowed’}, why?

The "Method Not Allowed" error you're seeing suggests that PATCH operations are not supported for the

/api/deployments/{id}

endpoint. This is actually expected behavior - in Prefect 3.x, deployment updates should be handled through the Python client rather than direct API calls. Instead of using a PATCH request, you should: 1. Use the Python client to update deployments:

python
from prefect import get_client

async with get_client() as client:
    # Fetch the deployment
    deployment = await client.read_deployment("deployment-id")
    
    # Update the deployment properties
    deployment.description = "New description"
    
    # Save the changes
    await client.update_deployment(deployment)

2. Or use the CLI:

bash
prefect deployment update my-deployment-name

If you're working on automation that requires HTTP API access, you might want to look into using the Python client in your automation code instead, as it provides a more stable and supported interface for deployment management. The direct REST API endpoints are primarily used by the Prefect UI and the Python client internally. For the most reliable experience, stick to the Python client or CLI for deployment management operations.

This JSONDecodeError typically occurs when the API request fails to return valid JSON, which often means there's either: 1. A connection issue to your Prefect server 2. An authentication problem 3. Invalid parameters in the request Here's how to troubleshoot: 1. First, verify your Prefect server is running and accessible:

python
from prefect import get_client

async with get_client() as client:
    # Test basic connectivity
    try:
        await client.hello()
    except Exception as e:
        print(f"Connection error: {e}")

2. Check your API URL and authentication:

python
import prefect.settings
print(prefect.settings.PREFECT_API_URL.value())

3. Try the read_deployments call with minimal parameters first:

python
async with get_client() as client:
    try:
        deployments = await client.read_deployments()
        print(f"Found {len(deployments)} deployments")
    except Exception as e:
        print(f"Error: {e}")

4. If that works, add parameters incrementally to identify which one might be causing the issue. Make sure: -

limit

is a positive integer -

offset

is a non-negative integer -

total_len

is actually defined in your code before using it -

DeploymentSort.NAME_ASC

is properly imported from

prefect.client.schemas

Can you try these steps and let me know: 1. Are you able to connect to the server at all? 2. Does the basic

read_deployments()

call without parameters work? 3. What version of Prefect are you using? (

prefect version

) This will help narrow down the exact cause of the JSONDecodeError.

@Marvin I’m doing this

async with get_client() as client:
    # Fetch the deployment
    deployment = await client.read_deployment("deployment-id")
    
    # Update the deployment properties
    deployment.description = "New description"
    
    # Save the changes
    await client.update_deployment(deployment)

and I get this prefect.exceptions.PrefectHTTPStatusError: Client error ‘422 Unprocessable Entity’

The 422 Unprocessable Entity error typically occurs when the deployment update request contains invalid data. There are a few common issues to check: 1. Make sure you're using the actual deployment ID, not the name. The ID should look like a UUID, for example: "12345678-1234-5678-1234-567812345678" 2. When updating a deployment, you need to use a DeploymentUpdate object. Here's the correct way:

python
from prefect import get_client
from prefect.client.schemas import DeploymentUpdate

async with get_client() as client:
    # First get the existing deployment
    deployment = await client.read_deployment("your-deployment-id")
    
    # Create a DeploymentUpdate object with your changes
    deployment_update = DeploymentUpdate(
        description="New description",
        # Keep other fields you want to preserve
        name=deployment.name,
        flow_id=deployment.flow_id,
    )
    
    # Update the deployment
    updated_deployment = await client.update_deployment(
        "your-deployment-id", 
        deployment_update
    )

3. If you only want to update specific fields, you can use a more targeted approach:

python
async with get_client() as client:
    updated_deployment = await client.update_deployment(
        "your-deployment-id",
        DeploymentUpdate(description="New description")
    )

The key differences from your original code are: - Using

DeploymentUpdate

instead of modifying the deployment object directly - Passing both the deployment ID and the update object to

update_deployment

If you're still getting a 422 error, can you share: 1. The complete error message (it should include details about which fields are invalid) 2. The actual deployment ID you're using (with sensitive parts redacted if necessary) This will help identify exactly what's causing the validation error.

Here's how you can query task runs with the specified tag and states:

python
from prefect.client import get_client
from prefect.client.schemas.filters import (
    TaskRunFilter,
    TaskRunFilterStateType,
    TaskRunFilterTags,
)

async def get_filtered_task_runs():
    client = get_client()
    
    # Create the filter
    task_filter = TaskRunFilter(
        tags=TaskRunFilterTags(all_=["my_tag"]),
        state_type=TaskRunFilterStateType(
            any_=[
                "RUNNING",
                "PENDING",
                "CANCELLING",
                "SCHEDULED"
            ]
        )
    )
    
    # Query the task runs
    task_runs = await client.read_task_runs(task_filter)
    return task_runs

# Use in an async context
task_runs = await get_filtered_task_runs()

A few important notes: 1. The code needs to run in an async context since the client methods are asynchronous 2. The

all_

parameter in

TaskRunFilterTags

ensures the task run has the specified tag 3. The

any_

parameter in

TaskRunFilterStateType

matches task runs that are in any of the specified states 4. State names should be uppercase strings as shown in the example You can iterate through the returned task runs to access their properties:

python
for task_run in task_runs:
    print(f"Task Run ID: {task_run.id}")
    print(f"State: {task_run.state.type}")
    print(f"Tags: {task_run.tags}")

The query will return all task runs that match both criteria (having the tag "my_tag" AND being in one of the specified states).

The issue is that we need to use

TaskRunFilterState

instead of

TaskRunFilterStateType

. Here's the correct way to combine the conditions with AND logic:

python
from prefect.client import get_client
from prefect.client.schemas.filters import (
    TaskRunFilter,
    TaskRunFilterTags,
    TaskRunFilterState,
)
from prefect.states import State, StateType

async def get_filtered_task_runs(tag: str):
    client = get_client()
    
    # Create the filter
    task_filter = TaskRunFilter(
        tags=TaskRunFilterTags(all_=[tag]),
        state=TaskRunFilterState(
            type=TaskRunFilterStateType(
                any_=[
                    StateType.RUNNING,
                    StateType.PENDING,
                    StateType.CANCELLING,
                    StateType.SCHEDULED,
                    StateType.PAUSED,
                ]
            )
        )
    )
    
    # Query the task runs
    task_runs = await client.read_task_runs(task_filter)
    return task_runs

The key differences in this corrected version are: 1. We use

TaskRunFilterState

instead of directly using

TaskRunFilterStateType

2. The state type filter is nested within the

state

parameter 3. We use

StateType

enum values directly instead of their string values This will properly return task runs that satisfy BOTH conditions: - Have the specified tag AND - Are in one of the specified states The filter conditions are combined with AND logic by default in the

TaskRunFilter

, so this will give you exactly what you're looking for - task runs that match both criteria simultaneously.

I see the issue now. You can simplify this by using just

state_type

directly in the

TaskRunFilter

instead of nesting it under

state

. Here's the corrected version:

python
task_filter = TaskRunFilter(
    tags=TaskRunFilterTags(all_=[tag]),
    state_type=TaskRunFilterStateType(
        any_=[
            StateType.RUNNING.value,
            StateType.PENDING.value,
            StateType.CANCELLING.value,
            StateType.SCHEDULED.value,
            StateType.PAUSED.value
        ]
    )
)

task_runs = await client.read_task_runs(
    task_run_filter=task_filter, 
    sort=TaskRunSort.NAME_ASC, 
    limit=limit,
    offset=offset
)

The key change is that we're using

state_type

directly in the

TaskRunFilter

instead of trying to nest it under a

state

parameter. This should resolve the validation error you're seeing while still maintaining the AND logic between the tags and state type conditions.

Ah, I see the issue now. According to the error,

TaskRunFilterState

requires a

name

field. Let's modify the filter to include both the state type and name:

python
task_filter = TaskRunFilter(
    tags=TaskRunFilterTags(all_=[tag]),
    state=TaskRunFilterState(
        name=TaskRunFilterStateName(
            any_=[
                StateType.RUNNING.value,
                StateType.PENDING.value,
                StateType.CANCELLING.value,
                StateType.SCHEDULED.value,
                StateType.PAUSED.value
            ]
        ),
        type=TaskRunFilterStateType(
            any_=[
                StateType.RUNNING.value,
                StateType.PENDING.value,
                StateType.CANCELLING.value,
                StateType.SCHEDULED.value,
                StateType.PAUSED.value
            ]
        )
    )
)

task_runs = await client.read_task_runs(
    task_run_filter=task_filter, 
    sort=TaskRunSort.NAME_ASC, 
    limit=limit,
    offset=offset
)

Or, as suggested in my previous response, you can use the simpler form by filtering directly on `state_type`:

python
task_filter = TaskRunFilter(
    tags=TaskRunFilterTags(all_=[tag]),
    state_type=TaskRunFilterStateType(
        any_=[
            StateType.RUNNING.value,
            StateType.PENDING.value,
            StateType.CANCELLING.value,
            StateType.SCHEDULED.value,
            StateType.PAUSED.value
        ]
    )
)

task_runs = await client.read_task_runs(
    task_run_filter=task_filter, 
    sort=TaskRunSort.NAME_ASC, 
    limit=limit,
    offset=offset
)

The second approach (using

state_type

directly) is simpler and should achieve what you want - finding task runs that have the specified tag AND are in any of the specified states.

For Prefect 3.2.14, try this approach:

python
from prefect.client import get_client
from prefect.client.schemas.filters import (
    TaskRunFilter,
    TaskRunFilterTags,
    TaskRunFilterState,
    TaskRunFilterStateName
)
from prefect.states import StateType

async def get_filtered_task_runs(tag: str):
    client = get_client()
    
    task_filter = TaskRunFilter(
        tags=TaskRunFilterTags(all_=[tag]),
        state=TaskRunFilterState(
            name=TaskRunFilterStateName(
                any_=[
                    "RUNNING",
                    "PENDING",
                    "CANCELLING",
                    "SCHEDULED",
                    "PAUSED"
                ]
            )
        )
    )
    
    task_runs = await client.read_task_runs(
        task_run_filter=task_filter, 
        sort=TaskRunSort.NAME_ASC, 
        limit=limit,
        offset=offset
    )
    return task_runs

The key changes are: 1. We're using just the state names directly as strings 2. We're using

TaskRunFilterStateName

instead of

TaskRunFilterStateType

3. The state names should be uppercase strings This should work with Prefect 3.2.14. Let me know if you're still seeing any errors with this approach.

To trigger a deployment using the Prefect CLI, you have two main options: 1. Quick run with default parameters:

bash
prefect deployment run 'flow-name/deployment-name'

2. Run with custom parameters:

bash
prefect deployment run 'flow-name/deployment-name' \
    --param param1=value1 \
    --param param2=value2 \
    --json '{"param3": "value3"}' \
    --scheduled-time "2024-01-01T00:00:00Z"

Key options you can use: - `--param`: Specify individual parameters - `--json`: Pass all parameters as a JSON string - `--scheduled-time`: Schedule the flow run for a specific time - `--params-file`: Load parameters from a JSON or YAML file - `--name`: Give the flow run a custom name For example:

bash
# Simple run
prefect deployment run 'etl-flow/daily-run'

# Run with parameters
prefect deployment run 'etl-flow/daily-run' \
    --param date="2024-01-01" \
    --param batch_size=1000

# Run with scheduled time
prefect deployment run 'etl-flow/daily-run' \
    --scheduled-time "2024-01-01T00:00:00Z"

# Run with parameters from file
prefect deployment run 'etl-flow/daily-run' \
    --params-file params.json

Note: Make sure to use the correct format for the deployment name:

flow-name/deployment-name

. You can see all available deployments by running

prefect deployment ls

.

Based on your error message and the research, this issue typically occurs in a few scenarios: 1. Race Condition: The task run was deleted or cleaned up before your code tried to access it. This can happen if: - The flow run was manually terminated - The task run was deleted while the flow was still running - There was a cleanup operation that removed the task run 2. Network/Configuration Issues: The Prefect server can't be reached or is misconfigured: - The URL in your error (`http://prefect_orion:4200`) suggests you're using a Docker setup - There might be connectivity issues between your client and the Prefect server 3. Timing Issues: If you're trying to access the task run information immediately after it's created or completed, there might be a race condition where the task run hasn't been fully registered in the database yet. To resolve this, you could try: 1. Add Retry Logic: If you're writing custom code to fetch task run information, add retry logic with a small delay:

python
from prefect.utilities.asyncutils import sync_compatible
from asyncio import sleep
from prefect import get_client

@sync_compatible
async def get_task_run_name_by_id(task_run_id, max_retries=3, delay_seconds=1):
    client = get_client()
    for attempt in range(max_retries):
        try:
            task_run = await client.read_task_run(task_run_id)
            return task_run.name
        except prefect.exceptions.ObjectNotFound:
            if attempt == max_retries - 1:
                raise
            await sleep(delay_seconds)

2. Verify Server Connection: Check that your Prefect server is running and accessible:

bash
prefect server status

3. Check Task Run Status: Before attempting to access the task run, verify it exists and is in the expected state using the Prefect UI or CLI. If you're still experiencing issues, it would be helpful to know: 1. Are you running Prefect in Docker? 2. Can you access the Prefect UI and see the task run there? 3. Does this happen consistently or intermittently?

Use

Variable.get

from the Prefect SDK. Basic read:

from prefect.variables import Variable

value = Variable.get("MY_VAR")  # raises if not set

Provide a default to avoid errors:

value = Variable.get("MY_VAR", default="fallback")

If you want to handle the missing case explicitly:

from prefect.exceptions import MissingVariableError

try:
    value = Variable.get("MY_VAR")
except MissingVariableError:
    value = "fallback"

Note: Variables are for non-sensitive config. Don’t use them for secrets. Docs: - Concepts: Variables in Prefect 3 - Python API: prefect.variables API

You can “get or create” it like this:

from prefect.variables import Variable
from prefect.exceptions import MissingVariableError

def get_or_create_variable(name: str, default):
    try:
        return Variable.get(name)
    except MissingVariableError:
        Variable.set(name, default, overwrite=False)
        return default

value = get_or_create_variable("XYZ", "WHAT")

Notes: -

Variable.get("XYZ")

raises

MissingVariableError

if it doesn’t exist; providing

default=

will not create it in Prefect. - Variable values must be JSON-serializable, not necessarily a JSON string. Python types like str, int, float, bool, None, dict, and list are fine. For example: -

Variable.set("XYZ", {"a": 1})

is valid. -

Variable.get("XYZ")

will return

{"a": 1}

. Docs: - Concepts: Variables in Prefect 3 - Python API: prefect.variables API

Right, my bad — in Prefect 3.x the exception you want is

ObjectNotFound

, not

MissingVariableError

(that was 2.x). Here’s a “get or create” that works in 3.x:

from prefect.variables import Variable
from prefect.exceptions import ObjectNotFound

def get_or_create_variable(name: str, default):
    try:
        return Variable.get(name)
    except ObjectNotFound:
        Variable.set(name, default, overwrite=False)
        return default

value = get_or_create_variable("XYZ", "WHAT")

Notes: - Variable values must be JSON-serializable (str, int, float, bool, None, list, dict, etc.). You don’t need to pass a JSON string; plain Python types are fine. -

default=

on

Variable.get

only returns the default; it does not create the variable in Prefect. Docs: - Concepts: Variables in Prefect 3 - API: prefect.variables API

Short answer: no.

Variable.get

and

Variable.set

are synchronous in Prefect 3.x, so your

get_or_create_variable

is a normal function and does not need

await

. Example in a flow:

from prefect import flow
from prefect.variables import Variable
from prefect.exceptions import ObjectNotFound

def get_or_create_variable(name: str, default):
    try:
        return Variable.get(name)
    except ObjectNotFound:
        Variable.set(name, default, overwrite=False)
        return default

@flow
def my_flow():
    v = get_or_create_variable("XYZ", "WHAT")
    print(v)

my_flow()

If you wrote an async flow and want to be extra polite to the event loop, you can offload the sync call:

import anyio

v = await anyio.to_thread.run_sync(get_or_create_variable, "XYZ", "WHAT")

Docs: - Concepts: Variables in Prefect 3 - API: prefect.variables API

That error means you’re trying to

await

something that isn’t awaitable. Prefect variables are synchronous. Do this (no await):

from prefect.variables import Variable

value = Variable.get("XYZ", default="WHAT")  # does not create it

If you want “get or create”:

from prefect.variables import Variable
from prefect.exceptions import ObjectNotFound

def get_or_create_variable(name: str, default):
    try:
        return Variable.get(name)
    except ObjectNotFound:
        Variable.set(name, default, overwrite=False)
        return default

v = get_or_create_variable("XYZ", "WHAT")  # no await

In an async flow, still no await:

from prefect import flow

@flow
async def my_flow():
    v = get_or_create_variable("XYZ", "WHAT")  # sync call is fine

If you accidentally wrote

async def get_or_create_variable(...):

, change it to a plain

def

. Docs: - Variables in Prefect 3 - prefect.variables API

You’re returning

None

because

Variable.get

doesn’t raise when missing in Prefect 3.x — it returns the

default

you pass, and since the default parameter itself defaults to

None

, you got exactly that. Your

except

never ran, so nothing was created. Fix by using a sentinel so you can detect “missing” and then create:

from prefect.variables import Variable

_MISSING = object()

def get_or_create_variable(name: str, default):
    value = Variable.get(name, default=_MISSING)
    if value is _MISSING:
        # Only create if it doesn't exist
        Variable.set(name, default, overwrite=False)
        return default
    return value

# usage
v = get_or_create_variable("run_spark", True)

Notes: - Prefer

overwrite=False

so you don’t clobber an existing value by accident. - Values must be JSON-serializable (bools like

True

are fine). - No

await

needed; these calls are synchronous. Verify it was created:

prefect variable get run_spark
# or
prefect variable inspect run_spark

Docs: - Concepts: Variables in Prefect 3 - API: prefect.variables API

@Marvin I’m running a task t = my_task.submit(somevars) I wait for it for finish. How can I check if it failed or not?

Great question. In Prefect 3.x, a submitted task returns a PrefectFuture. You have a few simple options to determine success/failure: 1) Easiest: use result() and catch failures

t = my_task.submit(somevars)

try:
    value = t.result()  # waits and raises if the task failed
    # success
except Exception as exc:
    # failed
    print(f"Task failed: {exc}")

2) If you’ve already waited with wait(), inspect the state

from prefect.client.schemas.objects import StateType

t = my_task.submit(somevars)
t.wait()  # blocks without raising

if t.state.type in (StateType.FAILED, StateType.CRASHED, StateType.CANCELLED):
    print("Task did not succeed:", t.state.message)
elif t.state.type == StateType.COMPLETED:
    value = t.result(raise_on_failure=False)  # safe to retrieve

3) Get a State directly at submit time

from prefect.client.schemas.objects import StateType

state = my_task.submit(somevars, return_state=True)
if state.type == StateType.COMPLETED:
    ...
else:
    print("Task did not succeed:", state.message)

Notes: -

result()

blocks and by default raises on failure; use

raise_on_failure=False

if you don’t want exceptions raised. -

wait()

just waits; it doesn’t raise. - Prefect futures don’t implement `.done()`/`.cancelled()` like

concurrent.futures

. Docs for more detail: - Prefect futures API - States concept

You’re seeing that because accessing the base-class

.state

property on a PrefectFuture is deprecated in Prefect 3.x. How to fix it: - Stop using

t.state

on the future. - Use one of these instead: 1) Preferred: use result() and catch failures

t = my_task.submit(somevars)
try:
    value = t.result()  # waits; raises if task failed
except Exception as exc:
    print(f"Task failed: {exc}")

2) If you need the State object, use get_state()

from prefect.client.schemas.objects import StateType

t = my_task.submit(somevars)
t.wait()  # blocks without raising
state = t.get_state()

if state.type == StateType.COMPLETED:
    value = t.result(raise_on_failure=False)
else:
    print(f"Task did not succeed ({state.type}): {state.message}")

3) Or request a State directly at submit time

from prefect.client.schemas.objects import StateType

state = my_task.submit(somevars, return_state=True)
if state.type != StateType.COMPLETED:
    print(f"Task did not succeed: {state.message}")

Only if you’re subclassing PrefectFuture: - Implement a

.state

property in your subclass or subclass

PrefectTaskRunFuture

instead (that’s what the warning text is addressing). Docs for reference: - Prefect futures API - States concept If you’ve already replaced

t.state

with

t.get_state()

(or the patterns above) and still see the warning, can you share a small snippet and your

prefect version

? I’ll help dig in.

You’re right — in Prefect 3.x there is no

get_state()

method on

PrefectTaskRunFuture

. Use the

state

property instead (optionally after

wait()

if you want to block until completion). You can also use

result()

to get the value or the

wait()

helper to manage multiple futures. Examples:

from prefect import flow, task
from prefect.futures import wait

@task
def add(x, y):
    return x + y

@flow
def my_flow():
    fut = add.submit(1, 2)

    # Option 1: block until the task finishes, then read its final state
    fut.wait()
    print(fut.state.type)         # e.g. "COMPLETED", "FAILED"
    print(fut.state.is_completed())
    print(fut.result())           # the task's return value

    # Option 2: work with many futures
    f2 = add.submit(10, 20)
    done, not_done = wait([fut, f2], timeout=5)
    for f in done:
        print("done:", f.state.name, f.result())
    for f in not_done:
        print("still running:", f.state.name)

my_flow()

Other useful bits: -

future.exception()

to access the exception (if failed) -

resolve_futures_to_states(...)

or

resolve_futures_to_results(...)

to recursively resolve futures in collections - Async variants:

wait_async()

/

result_async()

References: - Prefect futures API: API: prefect.futures - Running work concurrently: How-to: Run work concurrently - States concept: Concepts: States - Source (futures implementation): prefect/src/prefect/futures.py If you were following an example that used

get_state()

, it was likely written for Prefect 2.x or a different API surface. In 3.x, the

state

property is the supported way.

Great question. The “best” EC2 type depends on what you’re installing and where your flows actually run. Assuming Prefect 3.x: TL;DR - If you’re just running lightweight Prefect workers (connecting to Prefect Cloud or a remote work pool): a burstable general-purpose instance is perfect — e.g., t3a.small/medium (or t3.small/medium) with “unlimited” enabled. - If you’re self-hosting Prefect Server (API + services) in production: use a general-purpose instance, not compute-optimized. Start with 2–4 vCPU and 8–16 GB RAM — e.g., t3a.large/t3a.xlarge (unlimited) or m6a.large, and scale up as needed. Keep Postgres and Redis on managed services (RDS/ElastiCache) or separate instances. - If the same machine will execute your flow workloads: pick the family based on your workload, not Prefect itself: - CPU-bound: c6a/c7a - Memory-heavy: r6a/r7a - Mixed: m6a/m7a Why - Prefect Server and workers are mostly I/O-bound and lightweight; they don’t need compute-optimized (c6a) unless you’re also doing heavy computation on the same box. - Burstable (t3a/t3) is cost-effective and fine for orchestration, but enable “unlimited” or monitor CPU credits to avoid throttling. - For production self-hosting, separating Postgres and Redis is the biggest win for reliability and scale. Suggested starting points - Dev/small team, all-in-one box (server + dev worker): t3a.large (2 vCPU, 8 GB) + gp3 SSD (50–100 GB). Consider t3a.xlarge if you want more headroom. - Production server (with RDS Postgres + ElastiCache Redis): m6a.large or t3a.xlarge (unlimited). Scale vertically or run multiple server instances behind a load balancer as you grow. - Workers only (hybrid with Prefect Cloud or remote work pool): t3a.small/medium is typically enough; the heavy lifting should happen on your target infrastructure (ECS/EKS/Batch, etc.). Notes - If you’re considering Graviton (t4g/m7g/c7g), it can be cost-effective, but confirm your images/dependencies support arm64. Sticking to x86_64 (t3a/m6a/c6a) is the safest default. - Use gp3 SSD volumes and size for logs/artifacts if you store them locally; otherwise push logs/artifacts to S3. - Prefer managed Postgres (RDS) and Redis (ElastiCache) for production self-hosted setups. If you share a bit more about your setup (self-hosted or Prefect Cloud, expected concurrent flow runs, whether the instance will also run the actual tasks, and whether workloads are CPU or memory heavy), I can give you a concrete size recommendation. Helpful refs: - Self-hosted architecture overview: Prefect 3.x self-hosted guide - Docker Compose setup (what actually runs on the server): Docker Compose deployment - Worker concepts and sizing: Workers in Prefect

Short answer (Prefect 3.x): - If

another_task

fails,

my_task1

will not run. Because

wait_for

requires upstreams to be successful by default,

my_task1

will be prevented from starting. -

my_task2

will still run if you submit it with

wait_for=[allow_failure(task1)]

. The

allow_failure(...)

wrapper tells Prefect to proceed once

task1

reaches any terminal state (including failed or cancelled), instead of requiring it to be completed successfully. So in your example:

task1 = my_task1.submit(wait_for=[another_task])         # task1 won’t run if another_task fails
task2 = my_task2.submit(wait_for=[allow_failure(task1)]) # task2 will run after task1 reaches a terminal state

Notes: - Without

allow_failure(...)

,

task2

would be blocked (it wouldn’t run). -

allow_failure(...)

only affects scheduling. If you also pass a failed future as an input to a task, you’ll typically want to wrap that input with

allow_failure(...)

too so the failure doesn’t propagate as an exception when inputs are resolved. Docs and references: - Tasks (wait_for) - States in Prefect 3 - Cancellation and terminal states

Short answer: - If

another_task

fails,

my_task1.submit(wait_for=[another_task])

will not execute. Prefect will create a task run for

task1

, put it in a NotReady/Pending state due to the failed upstream, and it typically ends as Cancelled (i.e., “skipped because upstream failed”). -

allow_failure(task1)

inside

wait_for=[...]

does not change that.

wait_for

enforces that upstreams are COMPLETED; wrapping with

allow_failure

has no effect there. So

task2

will also not run and will typically be Cancelled due to upstream not completed. What to do if you want task2 to run even when task1 fails: - Do not put

task1

in

wait_for

. Instead, pass

task1

as an input parameter to

my_task2

wrapped with

allow_failure

. The

allow_failure

annotation is respected for parameter resolution and will let the downstream run while giving it access to the upstream’s state. Example:

task1 = my_task1.submit(wait_for=[another_task])

# This will NOT run if `another_task` fails:
# task2 = my_task2.submit(wait_for=[allow_failure(task1)])  # <-- avoid this

# Do this instead: pass task1 as a parameter wrapped with allow_failure
task2 = my_task2.submit(upstream_state=allow_failure(task1))

Inside

my_task2

, you can inspect

upstream_state

(a State-like object) to handle success/failure paths. References: - Tasks and dependencies (3.x): Tasks - allow_failure annotation: prefect.utilities.annotations.allow_failure - Engine handling of upstream failures (source): task_engine.py and UpstreamTaskError: exceptions.py

Thanks for the traceback — this is a known incompatibility between Prefect 3.6.4 and newer FastAPI releases. FastAPI recently started evaluating string type annotations at runtime; Prefect 3.6.4 has a dependency function with a string return annotation

"PrefectDBInterface"

that isn’t importable at runtime (only under TYPE_CHECKING), so FastAPI errors with NameError during app startup. What to do - Option 1 (recommended): Upgrade Prefect to the latest 3.x release, which includes a fix for this.

pip install -U prefect
  prefect server start

- Option 2 (stay on 3.6.4): Pin FastAPI to a compatible version (avoid 0.123+).

pip install "fastapi<0.123"
  prefect server start

How to verify your FastAPI version

python -c "import fastapi; print(fastapi.__version__)"
# or
pip show fastapi

If you’re running this in Docker/Compose (your logs show a service name like

prefect_orion

): - Add

fastapi<0.123

to your requirements.txt (or pip install line) in the image build - Or switch to the latest official Prefect 3.x image tag If you can share how you’re installing/running Prefect (Dockerfile/requirements, or exact pip commands), I can give precise steps to patch your setup.

In Prefect 3.x, each task call returns a PrefectFuture with a task_run_id property. For a list of task futures:

task_run_ids = [f.task_run_id for f in futures]
# If you need strings:
task_run_ids_str = [str(f.task_run_id) for f in futures]

Note on subflows: - Subflows return PrefectFlowRunFuture (not a task future). Those have flow_run_id instead of task_run_id. If your list may contain both task and subflow futures, you can separate them:

from prefect.futures import PrefectTaskRunFuture, PrefectFlowRunFuture

task_run_ids = [f.task_run_id for f in futures if isinstance(f, PrefectTaskRunFuture)]
flow_run_ids = [f.flow_run_id for f in futures if isinstance(f, PrefectFlowRunFuture)]

If you’re on Prefect 2.x: you likely used

.submit()

and would access

future.task_run.id

.

You’re right: in 3.6.4 the base class PrefectFuture.task_run_id is deprecated. The non-deprecated way is to use the concrete future types: - Task futures: PrefectTaskRunFuture.task_run_id - Subflow futures: PrefectFlowRunFuture.flow_run_id If your list only has task futures:

# inside a flow
task_run_ids = [f.task_run_id for f in futures]

If your list may contain subflow futures too, discriminate by type:

from prefect.futures import PrefectTaskRunFuture, PrefectFlowRunFuture

task_run_ids = [f.task_run_id for f in futures if isinstance(f, PrefectTaskRunFuture)]
flow_run_ids = [f.flow_run_id for f in futures if isinstance(f, PrefectFlowRunFuture)]

Notes: - In Prefect 3.x, calling a task inside a flow (e.g., my_task(...)) returns a PrefectTaskRunFuture; the above works whether you used direct calls or .delay(). - PrefectDistributedFuture and PrefectConcurrentFuture both inherit from PrefectTaskRunFuture, so .task_run_id works for them too. If you’re still seeing a deprecation warning, it’s usually from referencing the abstract PrefectFuture type; switching to the concrete types (or using isinstance checks) avoids it. Do you have any subflows in that list?

Short answer: there isn’t a dedicated JSON block in Prefect 3.x. Use the Secret block to store JSON-serializable data, and access it via .get(). Examples - Create/save a new “JSON” block

from prefect.blocks.system import Secret

payload = {
    "api_key": "abc123",
    "endpoint": "<https://example.com>",
    "retries": 3,
    "features": ["a", "b"]
}

Secret(value=payload).save("my-json", overwrite=True)

- Load and read its value

from prefect.blocks.system import Secret

block = Secret.load("my-json")
data = block.get()  # returns the deserialized value (dict/list/str/etc.)
print(data["api_key"])

- Update/write the block

from prefect.blocks.system import Secret

data = Secret.load("my-json").get()
data["retries"] = 5
Secret(value=data).save("my-json", overwrite=True)

- Use inside a flow/task

from prefect import flow
from prefect.blocks.system import Secret

@flow
def use_config():
    cfg = Secret.load("my-json").get()
    # use cfg...
    return cfg

Notes - Secret accepts any JSON-serializable value (dict, list, str, number, bool). - .get() returns the deserialized value; if the underlying value is a JSON string, it will be json.loads-ed for you. - If you previously used a JSON block in Prefect 2.x, the Secret block is the 3.x replacement.

There isn’t a built-in “JSON” block in Prefect 3.x. For non-sensitive JSON you have two good options: Option A (simplest): Variables - Best for small, non-secret config you want to read/write easily. - Write/set

from prefect.variables import Variable

value = {"api_url": "<https://example.com>", "retries": 3}
Variable.set("my_json_config", value, overwrite=True)

- Load/get

from prefect.variables import Variable

cfg = Variable.get("my_json_config")  # returns your dict/list/etc.
print(cfg["api_url"])

- Update

cfg = Variable.get("my_json_config")
cfg["retries"] = 5
Variable.set("my_json_config", cfg, overwrite=True)

Notes: - Values must be JSON-serializable. - You can add tags via

tags=["env:prod"]

if helpful. Option B (structured, non-secret): Create a lightweight custom Block - Best if you want a first-class Block (with schema/validation) that is not secret. - Define, save, load, update

from typing import Any, Dict
from prefect.blocks.core import Block

class JsonConfig(Block):
    data: Dict[str, Any]

# Save (creates the block type if needed)
JsonConfig(data={"api_url": "<https://example.com>", "retries": 3}).save("my-json", overwrite=True)

# Load
cfg_block = JsonConfig.load("my-json")
print(cfg_block.data["api_url"])

# Update and write back
cfg_block.data["retries"] = 5
cfg_block.save("my-json", overwrite=True)

How to choose: - Use Variable if you just need a simple key/value store for JSON and don’t need a custom schema. - Use a custom Block if you want a named block type that’s visible in the Blocks UI, supports validation/typing, and can grow with your config over time. If you tell me which route you prefer (Variable vs custom Block), I can tailor the snippet further (e.g., add validation, defaults, or common patterns).

✅ thought for 38.0 seconds

That custom Block is a good approach for non-sensitive JSON. A couple of tips to make reads/writes reliable: - Use a single normalized name and pass overwrite=True when updating an existing block - You can mutate the loaded block and save it back; no need to construct a new instance Example patterns Define the block

from typing import Any, Dict
from prefect.blocks.core import Block

class JsonConfig(Block):
    value: Dict[str, Any]

Create (first time) or update (subsequent)

name = dag_name.replace("_", "-")

# First-time create
JsonConfig(value={"foo": "bar"}).save(name, overwrite=True)

# Update later
cfg_block = JsonConfig.load(name)
cfg_block.value["foo"] = "baz"
cfg_block.save(name, overwrite=True)

If you already have a dict and just want to write it back, you can still do:

name = dag_name.replace("_", "-")
config = JsonConfig.load(name).value
# ...mutate config...
JsonConfig(value=config).save(name, overwrite=True)

But mutating the loaded block in-place (previous example) avoids recreating the instance. Load and read

name = dag_name.replace("_", "-")
config = JsonConfig.load(name).value

Create-if-missing helper

def upsert_json_config(name: str, updates: Dict[str, Any]):
    try:
        b = JsonConfig.load(name)
        b.value.update(updates)
    except Exception:
        # If the block or document doesn't exist yet
        b = JsonConfig(value=updates)
    b.save(name, overwrite=True)

Notes - The block type (JsonConfig) is registered on first save; a load before the first save will fail. - overwrite=True is required to update the existing block document under the same name. - If you later change the schema, you can load with validate=False, adjust fields, then save with overwrite=True.

Great question. Prefect didn’t “guess” anything—your data is stored under the field name you defined, and the UI just presents it in a simplified way. What actually happens under the hood: - Your block class defines a single field named

value

. When you call

.save(...)

, Prefect serializes the block by calling

Block.model_dump(...)

and includes the block’s fields by name. - That means the block document stored in the API has data like:

{
  "value": {
    "start-date": "",
    "end-date": ""
  }
}

- The UI builds a form from the block schema (which has one property:

value

) and renders the contents of that field. For a single-field block, it effectively “shows you the contents of value,” which can look like it’s not wrapped—but it is. You can verify this locally:

b = JsonConfig(value={"start-date": "", "end-date": ""})
print(b.model_dump())  # -> {'value': {'start-date': '', 'end-date': ''}}

And when you load:

loaded = JsonConfig.load(name)
print(loaded.value)  # -> {'start-date': '', 'end-date': ''}

Tips: - If you prefer a different top-level name in the UI (instead of “value”), rename the field:

class JsonConfig(Block):
      data: Dict[str, Any]

Then the stored document will be

{"data": {...}}

and the UI will render a “data” field. - For even nicer UI and validation, use typed fields instead of a generic dict:

class JsonConfig(Block):
      start_date: str
      end_date: str

The UI will render dedicated inputs for

start_date

and

end_date

. - When updating an existing block, remember to pass `overwrite=True`:

JsonConfig(value=config).save(name, overwrite=True)

Got it—you want to migrate existing “JSON” blocks to your new custom block class:

class JsonConfig(Block):
    value: Dict[str, Any]

Here’s a practical, safe migration pattern. It reads the old block’s data, writes it into your new block, and optionally replaces it under the same name. Option 1: Migrate one block by name (same name, overwrite)

from typing import Any, Dict
from prefect.blocks.core import Block
from prefect.client.orchestration import get_client
from prefect.utilities.asyncutils import sync_compatible

class JsonConfig(Block):
    value: Dict[str, Any]

@sync_compatible
async def migrate_json_block(old_type_slug: str, name: str):
    async with get_client() as client:
        # 1) Read the old block document by type slug and name
        old_doc = await client.read_block_document_by_name(
            name=name,
            block_type_slug=old_type_slug,
            include_secrets=True,  # safe even if not secret
        )

        # 2) Extract the JSON payload. Most JSON-like blocks store under "value"
        payload = old_doc.data.get("value", old_doc.data)

        # 3) Save to your new block, reusing the same name
        JsonConfig(value=payload).save(name, overwrite=True)
        return True

# Example usage:
# migrate_json_block("json", "my-config")    # if the old type slug was "json"
# migrate_json_block("secret", "my-config")  # if you were using Secret as JSON

Option 2: Create a new name first (safer), then cut over

@sync_compatible
async def migrate_json_block_new_name(old_type_slug: str, old_name: str, new_name: str):
    async with get_client() as client:
        old_doc = await client.read_block_document_by_name(
            name=old_name,
            block_type_slug=old_type_slug,
            include_secrets=True,
        )
        payload = old_doc.data.get("value", old_doc.data)
        JsonConfig(value=payload).save(new_name, overwrite=True)
        return True

# Example: migrate_json_block_new_name("json", "my-config", "my-config-v2")

Option 3: Batch migrate all old JSON blocks

@sync_compatible
async def batch_migrate(old_type_slug: str, dry_run: bool = True):
    async with get_client() as client:
        all_docs = await client.read_block_documents()
        old_docs = [b for b in all_docs if b.block_type.slug == old_type_slug]

        migrated = []
        for doc in old_docs:
            payload = doc.data.get("value", doc.data)
            if dry_run:
                print(f"[DRY RUN] Would migrate: {old_type_slug}/{doc.name}")
            else:
                JsonConfig(value=payload).save(doc.name, overwrite=True)
                print(f"Migrated: {doc.name}")
            migrated.append(doc.name)
        return migrated

# Preview:
# batch_migrate("json", dry_run=True)
# Then execute:
# batch_migrate("json", dry_run=False)

Notes and tips - Finding the old type slug: If you used the Prefect 2.x JSON block, its block type slug was typically “json”. If you used Secret to hold JSON, use “secret”. - Where is the data? Most JSON-like blocks store the actual value under the key “value” in the block document. The code above falls back to the entire document data if needed. - Overwrite behavior: Passing overwrite=True lets you replace the existing block document under the same name. If you want a safer cutover, save under a new name first. - Validation changes: If your old block had a different schema and you hit validation errors when loading, try loading with

validate=False

, massage the data, then save. - Optional cleanup: After verifying the new block works, you can delete the old block via UI or CLI: - CLI:

prefect blocks delete <type-slug>/<name>

If you share the exact old block type/slug you’re migrating from, I can tailor the snippet further.

That warning means you called a sync_compatible-wrapped async function without awaiting it while already in an async context. sync_compatible lets the same function be used in both contexts: - In normal (sync) code: call it directly (no await) - In async code (e.g., inside an async def or a notebook/REPL with a running event loop): await it Fix your call site based on where you’re running it: - From a regular script or REPL:

migrate_json_block("json", "my-config")

- From inside an async function or an environment with a running event loop (Jupyter, some frameworks, or an async Prefect flow/task):

await migrate_json_block("json", "my-config")

Alternatively, remove the decorator and drive it explicitly with asyncio:

import asyncio

async def migrate_json_block(old_type_slug: str, name: str):
    async with get_client() as client:
        old_doc = await client.read_block_document_by_name(
            name=name,
            block_type_slug=old_type_slug,
            include_secrets=True,
        )
        payload = old_doc.data.get("value", old_doc.data)
        JsonConfig(value=payload).save(name, overwrite=True)

# Sync entrypoint
if __name__ == "__main__":
    asyncio.run(migrate_json_block("json", "my-config"))

Quick sanity check for the payload extraction:

payload = old_doc.data["value"] if "value" in old_doc.data else old_doc.data
JsonConfig(value=payload).save(name, overwrite=True)

If you share where you’re calling it from (script, notebook, inside a flow/task), I can suggest the cleanest version for that context.

Yes—you can horizontally scale Prefect Server behind your ALB to distribute API and websocket load, as long as all instances share the same database and encryption key and you configure a couple of things carefully. What to configure/check Database - Use the same Postgres RDS for all instances. - Set the same connection URL on every instance: - PREFECT_SERVER_DATABASE_CONNECTION_URL (or PREFECT_API_DATABASE_CONNECTION_URL) - Example: postgresql+asyncpg://user:pass@host:5432/dbname - Ensure your DB max_connections can handle pool_size × number_of_instances. - Default per-instance pool_size is modest, but it adds up when you scale. Encryption key (critical) - All instances must use the same encryption key so they can decrypt stored secrets and block data: - PREFECT_SERVER_ENCRYPTION_KEY - Do not rely on auto-generation on a new instance; explicitly set the same key across all. Migrations - Avoid running migrations concurrently across instances. - Recommended: - Run migrations once (e.g., start a single instance with PREFECT_SERVER_DATABASE_MIGRATE_ON_START=true), then - Set PREFECT_SERVER_DATABASE_MIGRATE_ON_START=false on all instances thereafter. - Alternatively, keep it false everywhere and run migrations manually as a one-time step before scaling. Load balancer (ALB) - Health checks: point to /health on your Prefect API port. - Websockets: enable stickiness on the target group so long-lived websocket connections aren’t bounced between instances. - Align timeouts: - ALB idle timeout should be less than or equal to PREFECT_SERVER_API_KEEPALIVE_TIMEOUT (or increase keepalive to exceed ALB idle timeout). Agent(s) and workers - You can run multiple agents/workers; they’ll compete for work from the same server and are safe to scale horizontally. - Agents don’t need stickiness; only the server’s websocket/API do. Scheduler and background services - Prefect’s server services (e.g., scheduler) are designed to be idempotent; running them on multiple instances is safe. - Still, ensure all instances point to the same DB and share the same encryption key as above. Other consistency settings - Keep these consistent (or intentionally set) across instances: - PREFECT_SERVER_API_AUTH_STRING (if you use it) - PREFECT_SERVER_API_CSRF_PROTECTION_ENABLED - PREFECT_SERVER_API_KEEPALIVE_TIMEOUT - Host/port/base path can vary per instance, but behind ALB they typically match. SQLite caveat - Do not use SQLite for multi-instance. You’re on Postgres RDS already, which is correct for scaling. Quick checklist - [ ] All instances use the same Postgres RDS URL - [ ] PREFECT_SERVER_ENCRYPTION_KEY is identical on all instances - [ ] Run DB migrations once; disable on others (PREFECT_SERVER_DATABASE_MIGRATE_ON_START=false) - [ ] ALB stickiness enabled for websocket stability - [ ] ALB health check to /health - [ ] DB connection limits sized for your instance count If you want, share your current env var setup (scrub secrets), and I’ll double-check it for scale-out readiness.

Short answer: No. Server-side encryption of block document data is mandatory in Prefect 3.x and cannot be disabled. Key points - Prefect encrypts block document data using a Fernet key. There is no config to turn this off. - How the key is determined: - If PREFECT_SERVER_ENCRYPTION_KEY is set, that key is used. - Else, Prefect looks for a key stored in the database Configuration table. - If none exists (first startup), Prefect auto-generates a new Fernet key and stores it in the DB. - Implications for multi-instance: - All server instances must use the same key (either the same env var or the same DB so they read the same stored key). If instances use different keys, they will fail to decrypt each other’s data. - Backups/migrations: - If you move to a new database, bring the key with you (set PREFECT_SERVER_ENCRYPTION_KEY to the original value) or existing encrypted block data will be unreadable. If you want, I can help you extract the current key reference (from env or DB) and set it consistently across your autoscaling instances.

Yes. Since you didn’t set PREFECT_SERVER_ENCRYPTION_KEY, Prefect generated and stored the key in your database’s configuration table. You can extract it and set it consistently across instances. Where it lives - Table: configuration - Row key: ENCRYPTION_KEY - JSON value: {"fernet_key": "<base64-encoded-fernet-key>"} Ways to read it Option A: Direct SQL (Postgres)

-- Fernet key only
SELECT value->>'fernet_key' AS fernet_key
FROM configuration
WHERE key = 'ENCRYPTION_KEY';

-- Full row
SELECT id, key, value, created, updated
FROM configuration
WHERE key = 'ENCRYPTION_KEY';

Option B: Python via Prefect’s DB models (idiomatic)

import asyncio
from prefect.server.database import provide_database_interface
from prefect.server.models import configuration as config_models

async def read_prefect_encryption_key():
    db = provide_database_interface()
    async with db.session_context() as session:
        cfg = await config_models.read_configuration(session=session, key="ENCRYPTION_KEY")
        return cfg.value["fernet_key"] if cfg else None

# Usage
key = asyncio.run(read_prefect_encryption_key())
print(key)  # base64 Fernet key

Option C: Minimal SQLAlchemy (if you just have the DB URL)

import asyncio
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
import uuid

class Base(DeclarativeBase): pass

class Configuration(Base):
    __tablename__ = "configuration"
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
    key: Mapped[str] = mapped_column(unique=True, index=True)
    value: Mapped[dict] = mapped_column(sa.JSON)

async def get_key(db_url: str):
    engine = create_async_engine(db_url, echo=False)
    async with AsyncSession(engine) as session:
        result = await session.execute(sa.select(Configuration).where(Configuration.key == "ENCRYPTION_KEY"))
        row = result.scalar_one_or_none()
        return row.value["fernet_key"] if row else None

# Usage
# key = asyncio.run(get_key(os.environ["PREFECT_API_DATABASE_CONNECTION_URL"]))

After you retrieve it - Set it as an env var on every server instance: - Linux shell: export PREFECT_SERVER_ENCRYPTION_KEY='the_fernet_key' - In systemd, container env, or your ASG launch template/user-data - Best practice: store it in a secrets manager (AWS Secrets Manager/SSM Parameter Store) and inject it at boot. Security notes - Treat this key like a secret; it decrypts all encrypted block data. - Don’t log it or commit it to version control. - Prefer injecting via secret manager or a protected file (0600) rather than inline shell exports. If you share how your instances are launched (user data, ASG launch template, container task def), I can suggest the cleanest way to inject PREFECT_SERVER_ENCRYPTION_KEY there.

Set it only on your Prefect Server (Orion) containers. Agents/workers do not need the encryption key. Why: - The server encrypts data before storing it and decrypts it when serving API responses. Clients (agents, workers, flow runs) receive decrypted data via the API and do not perform decryption themselves. - Therefore, only processes running the Prefect Server need PREFECT_SERVER_ENCRYPTION_KEY. Checklist - Orion/server containers behind your ALB: set the same PREFECT_SERVER_ENCRYPTION_KEY on all of them. - Agent/worker containers: not required (safe but unnecessary to set). - If a container runs both server and agent in one process: set it there. Tip: Store the key in AWS Secrets Manager/SSM and inject it into just the server tasks. If you ever see “InvalidToken” or decryption errors in server logs when accessing blocks, it usually means one server instance has a different/missing key.

Short answer - Yes, you should keep the key stable and backed up even if you don’t plan to store secrets. It’s used to encrypt ALL block document data; losing or changing it arbitrarily will make existing blocks unreadable. - If you set PREFECT_SERVER_ENCRYPTION_KEY now, Prefect will use it on the next server start. Set it to the exact key that’s already stored in your database, not a new value. - Changing the key later (rotation) is not built-in and requires a careful re-encryption process with downtime. Generally, avoid rotating unless you have a strong reason. Details Do I need to keep the key safe if I never store secrets? - Prefect encrypts block document data regardless of whether it’s sensitive. If you have any saved Blocks (including your custom JSON config blocks), they’re encrypted at rest. - If the key changes or is lost, Prefect won’t be able to decrypt those blocks, and loads/reads of those blocks will fail. Even if the contents aren’t sensitive, you’d lose access to them. If I set PREFECT_SERVER_ENCRYPTION_KEY now, will Prefect pick it up? - Yes. The server reads the env var at startup and prefers it over the DB-stored key. You must restart your Prefect Server containers for it to take effect. - Important: Set it to the existing key that Prefect generated earlier, otherwise previously encrypted block data will become unreadable. - To fetch the current key from Postgres:

SELECT value->>'fernet_key' AS fernet_key
  FROM configuration
  WHERE key = 'ENCRYPTION_KEY';

Then set:

export PREFECT_SERVER_ENCRYPTION_KEY='<that_fernet_key>'

Apply this to all Prefect Server instances (Orion containers) and restart them. Do I need to set it on agents/workers? - No. Only the Prefect Server (Orion) needs the encryption key. How do I change/rotate the key later? - There’s no one-click rotation. To rotate safely you need a maintenance window and a re-encryption process. High-level approach: 1) Back up your database. 2) While running with the old key, export all block documents’ decrypted data (e.g., via the API/Prefect client or a script). 3) Stop the server. 4) Set PREFECT_SERVER_ENCRYPTION_KEY to the new key on all server instances and start the server. 5) Re-save each block document (write back the same data) so it’s re-encrypted with the new key. 6) Validate loads work; keep the old key somewhere secure until you’ve verified everything. - If you don’t need rotation for compliance, the best practice is to pick a strong key now, store it in a secret manager (e.g., AWS Secrets Manager/SSM), and keep it stable. Recommended path for you - Extract the current key from the database (query above). - Set PREFECT_SERVER_ENCRYPTION_KEY to that exact value on all Orion/server instances (not agents). - Store it in AWS Secrets Manager/SSM and inject it at boot. - Avoid changing the key unless you must; if needed, plan a re-encryption migration as described.

Yes, you can safely delete TELEMETRY_SESSION. It’s created and used by the server’s telemetry service to track an anonymous session (UUID + start timestamp) for periodic heartbeat analytics. What happens if you delete it - The next time the telemetry service runs, it will recreate TELEMETRY_SESSION automatically with a new session ID and timestamp. - No data integrity impact; it only resets the telemetry session. If you don’t want it recreated - Disable server analytics: - PREFECT_SERVER_ANALYTICS_ENABLED=false - With analytics disabled, the telemetry service won’t run and the entry won’t be recreated. Summary - Safe to delete: yes. - Will be recreated: yes, unless analytics are disabled.

Short answer: An empty value is treated as “unset,” so Prefect will ignore it and fall back to the key in the database (or generate and store a new one if none exists). It does not disable encryption. Details - If PREFECT_SERVER_ENCRYPTION_KEY == "" (empty string), Prefect treats it as falsy and does not use it. It will: - Use the existing key from the configuration table (ENCRYPTION_KEY), or - Generate a new Fernet key and store it in the configuration table if none exists (first-time startup). - If you set a non-empty but invalid string (e.g., a space), Prefect will try to create a Fernet key from it and raise an error. Don’t do this. Recommendations - Prefer either: - Not setting the env var at all and relying on the DB-stored key, or - Setting it to the exact, non-empty Fernet key you retrieved from the DB so it’s consistent across all server instances. - In multi-instance setups, set the same non-empty key on all server (Orion) instances. - Don’t set it to an empty string in Compose/K8s/ECS; omit the env var instead: - Shell: run

unset PREFECT_SERVER_ENCRYPTION_KEY

rather than

export PREFECT_SERVER_ENCRYPTION_KEY=

- Docker Compose: remove the line rather than

PREFECT_SERVER_ENCRYPTION_KEY=

- Kubernetes: omit the env entry, or source it from a secret with a real value If your DB doesn’t have a key yet and multiple servers start at once with empty/unset keys, you can have a race to generate the key. Seed it by starting one server first (or set the env var explicitly) to avoid that race.

Summary - PREFECT_SERVER_DATABASE_MIGRATE_ON_START controls whether the server attempts to migrate the database schema during startup. Default is true. - If true, the server will try to run migrations on each startup. Prefect guards “once per app instance” within a process, but not across different processes/hosts. Multi-instance implications - With multiple server instances starting at the same time and migrate_on_start=true everywhere: - Each instance will attempt to run Alembic migrations when it starts. - Alembic tracks applied revisions in the database (alembic_version table) and migrations are idempotent, so you won’t corrupt the schema. - However, you can see lock contention and startup delays while instances serialize on migration attempts (especially problematic with SQLite; with Postgres it’s usually just extra noise/delay). Best practice for autoscaling/multi-instance - Run migrations once, then disable migrate_on_start on the API instances. - Run a dedicated one-off migration step before your API replicas start (e.g., an init job or a separate “migrate” task/container that runs prefect server database upgrade -y). - Set PREFECT_SERVER_DATABASE_MIGRATE_ON_START=false on all API/server instances. - This avoids race conditions and speeds up startup for scaled replicas. If you keep it true everywhere - It will still work (migrations are idempotent), but: - Expect unnecessary repeated upgrade checks at every startup. - Potential transient lock contention and longer startup times when instances come up together. - Not recommended as you scale. Quick checklist - [ ] Use Postgres (not SQLite) for multi-instance - [ ] Add a single migration job (prefect server database upgrade -y) to your deployment pipeline - [ ] Set PREFECT_SERVER_DATABASE_MIGRATE_ON_START=false on your server instances - [ ] Bring up N replicas behind your ALB after migration completes If you share your current deployment method (Docker Compose, ECS, K8s), I can provide a concrete migration job example for it.

Great question. Think of migrations as “bring the database schema up (or down) to match the Prefect server code you’re about to run.” Here are the events where you should run them, and whether to do it before or after. When to run migrations - New/empty database (e.g., new RDS, switching DBs) - Action: Run upgrade BEFORE starting any Prefect server instances. - Command:

prefect server database upgrade -y

- Upgrading Prefect server version (new Docker image/package) - Action: Run upgrade BEFORE starting the new server version (ideally as a one-off job). This ensures the DB schema matches the new code. - Command:

prefect server database upgrade -y

- Note: For multi-instance, run this once, then start/roll out the new replicas with

PREFECT_SERVER_DATABASE_MIGRATE_ON_START=false

. - Restoring a database backup that’s older than your target server version - Action: AFTER the restore, run upgrade BEFORE starting the server. - Command:

prefect server database upgrade -y

- Downgrading Prefect server version (not generally recommended) - Action: BEFORE starting the older server, either restore a DB backup taken at that older version, or run an explicit downgrade to the matching revision (if supported). - Commands:

prefect server database downgrade -r <target_revision>
    # or stamp if you know the DB already matches:
    prefect server database stamp -r <revision>

- Safer: restore a DB snapshot that matches the target version. Events that do NOT require a migration - Scaling the number of server/agent instances up or down (same Prefect version) - Changing environment variables (including encryption key) — these aren’t schema changes - Creating/updating Blocks, Deployments, Work Pools, etc. - Upgrading only agents/workers or your flow code/client libraries How to check if a migration is needed - Use a dry run to see pending migrations:

prefect server database upgrade --dry-run

If nothing is printed, you’re already at head. Recommended pattern for autoscaling/multi-instance - Run a single migration step (e.g., an init job/one-off task) as part of your deployment pipeline:

prefect server database upgrade -y

- Start your API replicas with

PREFECT_SERVER_DATABASE_MIGRATE_ON_START=false

. If you share your deployment method (Compose, ECS, EKS), I can suggest an exact “migration job + replicas” layout for it.

Short answer - You don’t need to scale down to 1 to run migrations. The cleanest approach is to run a one-off migration job, then roll your server instances (with migrate_on_start=false). No API downtime and no drain needed. - If you do choose to scale down, you don’t have to wait for zero running deployments, but schedule a short maintenance window if you want to avoid any transient API/logging hiccups. Two safe upgrade patterns A) Preferred: one-off migration job (no scale-down) 1) Pre-check (optional):

prefect server database upgrade --dry-run

If nothing prints, there’s nothing to migrate. 2) Run the migration once (with the new Prefect image/version):

prefect server database upgrade -y

- Use the same PREFECT_SERVER_DATABASE_CONNECTION_URL and PREFECT_SERVER_ENCRYPTION_KEY as your servers. - This can be a short-lived container/task (ECS one-off task, K8s Job, or a script on an instance). 3) Roll your autoscaling group to the new image/version with: - PREFECT_SERVER_DATABASE_MIGRATE_ON_START=false (on all server instances) - Use a rolling update so ALB keeps at least one target healthy at all times. Why this is better - No need to scale down. - No race conditions between instances. - Minimal to no API interruption. B) If you still want to scale down to 1 and let the server do migrations at boot 1) Scale down to 1 instance (or put all but one instance in Standby/detach from target group), keeping one server instance healthy behind the ALB. 2) Redeploy that 1 instance with the new Prefect version and: - PREFECT_SERVER_DATABASE_MIGRATE_ON_START=true (only for this boot) - It will migrate the DB on startup. 3) After it’s healthy and the DB is upgraded, switch that instance to: - PREFECT_SERVER_DATABASE_MIGRATE_ON_START=false 4) Scale back up (rolling). Do I need to wait until no deployments are running? - Not strictly required. - Prefect’s Alembic migrations are idempotent and usually quick on Postgres. - Running flows continue wherever they’re executing; scheduling will catch up if the scheduler is briefly unavailable. - That said, if you stop all server instances (or your ALB routes no traffic) you’ll temporarily lose API/UI and log delivery; long-running tasks won’t stop, but log/heartbeat requests may fail during the outage. - To minimize impact: - Prefer the one-off migration job so the API can remain up. - If you must bounce the API, aim for a brief maintenance window or pause new runs. - Optionally pause deployments to avoid new runs starting during the swap; the scheduler will resync once the API is back. Extra safeguards - Back up your database before any migration. - Verify the key env vars are identical on all server instances: - PREFECT_SERVER_DATABASE_CONNECTION_URL - PREFECT_SERVER_ENCRYPTION_KEY - Use the dry-run first:

prefect server database upgrade --dry-run

If you share your deployment method (ECS/K8s/EC2 systemd), I can give a concrete one-off migration task/job example tailored to it.

You’re right to be cautious. If you migrate the database while older API servers are still serving traffic, those older servers can break if the migration includes a non‑backward‑compatible change (e.g., column rename/drop, constraint change). Prefect tries to keep migrations mostly additive, but there is no blanket guarantee of backward compatibility across all versions. Practical guidance - Safest pattern (brief maintenance window) - Take old API servers out of the load balancer (or scale to 0) so they stop serving traffic. - Run the migration once:

prefect server database upgrade -y

- Roll out the new server version (with PREFECT_SERVER_DATABASE_MIGRATE_ON_START=false). - Reattach/scale up. - Impact: brief API/UI downtime. Running flow runs continue; some API calls (heartbeats/logs) may fail transiently and retry once API is back. - Zero/near‑zero downtime pattern (only if you’ve verified compatibility) - In a staging environment, confirm the migration is backward compatible with your current production server version. - Run a one‑off migration job while old servers are still up (keep the window short). - Immediately roll the API servers to the new version (migrate_on_start=false). - Monitor for 5xx errors. If you see schema errors, revert quickly and perform the maintenance-window method. - “Scale down to 1 then migrate” pattern - It reduces exposure, but one remaining old instance could still error during the migration window. If you use this, do the scale-down and remove that instance from the LB first, migrate, then roll the instance to the new version and bring it back. Do I need to wait for no running deployments? - Not required. Running tasks will continue on agents/workers. If the API is down briefly, some log/heartbeat calls may fail but will generally retry. If you want zero impact on logs/UI, schedule a short maintenance window. Summary - Yes, old server instances can break against a newly migrated schema. - Use a short maintenance window (detach/scale down API servers, migrate, then roll up the new version) for the safest upgrade. - If you must avoid downtime, first verify the specific migration is backward compatible in staging; then keep the overlap window short and roll immediately.

Only on Orion (the Prefect Server). Agents/workers don’t run migrations and don’t need (or use)

PREFECT_SERVER_DATABASE_MIGRATE_ON_START

. Recommendations: - Orion/server containers: set it as needed (ideally

false

and run a one‑off

prefect server database upgrade -y

before rolling). - Agent/worker containers: do not set it; it has no effect. - If you run server and agent in the same container/process, the setting applies only to the server process.

Short answer: Not necessarily. It depends on where the flow code is actually running. What happens when you scale down instances that host both Orion (server) and a worker/agent: - If your worker runs flows on the same instance (process/subprocess worker, or Docker containers started on that host): - Terminating the instance will kill those processes/containers. Those flow runs will be lost (marked crashed/failed unless they have retries). - If your worker launches flows onto external infrastructure (Kubernetes Jobs, ECS tasks, remote Docker hosts): - Those jobs/pods/tasks continue running even if the worker instance goes away. The worker is only needed to pick up new work; running flows continue and report state/logs directly to the Prefect server. - Scaling down Orion (server) replicas: - As long as at least one server instance remains healthy behind your ALB, running flows keep communicating normally. - If all server instances go down temporarily, flows will keep running but API calls (logs/heartbeats/state updates) may temporarily fail and retry. Extended outages can lead to incomplete state reporting. Recommended “drain” steps before terminating instances that host active runs - If the worker is local/process/docker-on-host: - Pause the work pool(s) so no new work is picked up:

prefect work-pool pause <pool-name>

- Wait for running flow runs to finish (UI: filter by State=Running; CLI:

prefect flow-run ls --state-name Running

and check relevant deployments). - Then terminate the instance(s). - If the worker uses external infra (K8s/ECS): - You can stop the worker without killing running jobs; they’ll continue. Still pause the pool to avoid new work landing on disappearing capacity. Operational best practices - Separate concerns: Run Orion on autoscaled instances; run workers that launch to external infra (K8s/ECS) so scaling the ASG does not kill runs. - For rolling updates/scale downs: - Pause pools to prevent new dispatch. - Let in-flight local runs complete or cancel them explicitly. - Deregister instances from the ALB first (server side) so traffic drains; keep at least one server up. - For ASG, consider lifecycle hooks to trigger a “drain” script (pause pools, stop worker polling, wait, then terminate). If you tell me which worker type you’re using (process, docker, kubernetes, ecs, etc.), I can give exact drain/scale procedures for that setup.

Short answer: There isn’t a supported setting or env var to disable that toast globally in the Orion UI. The UI shows error toasts from component-level error handlers; there’s no global “turn off error toasts” switch. What you can do - If you control the UI build (self-hosting the Prefect UI code), you can patch it: - Add filtering in component onError callbacks that call toast.error(...) - Or monkey‑patch the Sonner toast instance at app bootstrap to drop messages you don’t want - If you’re just running the stock Prefect UI image, there’s no built-in way to suppress that specific toast without forking the UI image. Pointers if you can rebuild the UI - Error toasts are typically triggered in onError handlers across components (e.g., block create/update pages, deployments, work pools). - The toast component uses Sonner. You can wrap/patch it to filter messages (e.g., if the message contains “Cloud”, skip). - There’s no VITE_* or PREFECT_* env var that toggles that behavior. If you can share where/when the toast appears (which page/action), I can point you to the exact file/function in the UI code where it’s raised and show the minimal patch to remove it.

In Prefect 3.x, you have a few choices depending on what you need. If you just need metadata (including the state at context entry): - Use get_run_context() inside the task. Note: the state fields here are a snapshot from when the task context was created, not live-refreshed.

from prefect import task
from prefect.context import get_run_context

@task
def my_task():
    ctx = get_run_context()
    # Snapshot at start of task context
    print(ctx.task_run.state_type)  # e.g., StateType.RUNNING
    print(ctx.task_run.state_name)  # "Running"

If you need the current state from the server (live): - Query the server using the orchestration client and your own task_run_id from context. Sync task:

from prefect import task
from prefect.context import get_run_context
from prefect.client.orchestration import get_client

@task
def my_task():
    task_run_id = get_run_context().task_run.id
    with get_client(sync_client=True) as client:
        states = client.read_task_run_states(task_run_id)
    current = states[-1] if states else None
    if current:
        print(current.type, current.name, current.message)

Async task:

from prefect import task
from prefect.context import get_run_context
from prefect.client.orchestration import get_client

@task
async def my_task():
    task_run_id = get_run_context().task_run.id
    async with get_client() as client:
        states = await client.read_task_run_states(task_run_id)
    current = states[-1] if states else None
    if current:
        print(current.type, current.name, current.message)

Notes and caveats: - prefect.runtime.task_run exposes IDs, names, tags, etc., but not state. - Setting state from within a task is advanced and can interfere with orchestration; only do it if you know why: -

client.set_task_run_state(task_run_id, SomeState(...))

- In most cases, the snapshot from get_run_context() is sufficient to know you’re running; use a server query only if you truly need the latest state mid-execution.

You don’t need the full history to get the current state. Use read_task_run(...) instead of read_task_run_states(...). The TaskRun object includes the current/latest state. Async:

from prefect.client.orchestration import get_client

async def get_current_state(task_run_id):
    async with get_client() as client:
        tr = await client.read_task_run(task_run_id)
        return tr.state  # current/latest State

Sync:

from prefect.client.orchestration import get_client

def get_current_state(task_run_id):
    with get_client(sync_client=True) as client:
        tr = client.read_task_run(task_run_id)
        return tr.state

Notes: - read_task_run_states(task_run_id) returns the entire state history (oldest → newest) for auditing; latest is states[-1]. - read_task_run(task_run_id) is the efficient way to get just the current state (plus summary fields like state_id, state_type, state_name).

Short version: - state is the full State object for the task’s current state (rich details). - state_type is just the enum summary of that state (quick to check/filter). Details: - state (prefect.states.State) - Rich object describing the current state. - Includes: type (enum), name (string), message, timestamp, state_details (e.g., run count, pause/resume info), and optionally data. - Use when you need context like why it failed, when it changed, extra details, etc. - state_type (prefect.client.schemas.objects.StateType) - A denormalized enum copied onto the TaskRun for convenience and fast querying (e.g., RUNNING, COMPLETED, FAILED, CRASHED, PENDING, SCHEDULED). - Use when you just need to know “what status is it in?” for flags, filters, or UI badges. Also: - state_name is the human-friendly string (e.g., "Running", "Completed"), useful for display. - state_id is the current state row’s ID in the database. - Either can be None if a run has not had state set yet (rare, typically you’ll see at least PENDING/SCHEDULED). Example:

from prefect.client.orchestration import get_client

async def inspect(task_run_id):
    async with get_client() as client:
        tr = await client.read_task_run(task_run_id)

        # Quick status check
        print(tr.state_type)   # e.g., StateType.RUNNING
        print(tr.state_name)   # "Running"

        # Detailed info
        s = tr.state
        if s:
            print(s.type, s.name, s.message, s.timestamp)
            # s.state_details, s.data, etc.

Yes—use the client to fetch all task runs for flow run Y, then set each task’s state. Basic pattern (async flow/task):

from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterFlowRunId
from prefect.states import Cancelled  # or Failed(), etc.

async def set_all_tasks_state(flow_run_id: UUID, new_state):
    async with get_client() as client:
        task_runs = await client.read_task_runs(
            task_run_filter=TaskRunFilter(
                flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id])
            )
        )
        results = []
        for tr in task_runs:
            # Optional: skip terminal tasks
            if tr.state_type in {"COMPLETED", "FAILED", "CANCELLED", "CRASHED"}:
                continue
            res = await client.set_task_run_state(tr.id, new_state)  # add force=True only if you must
            results.append((tr.id, res.accepted))
        return results

# Example usage:
# await set_all_tasks_state(flow_run_id_y, Cancelled(message="Cancelled by controller flow X"))

Sync variant:

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterFlowRunId
from prefect.states import Cancelled

def set_all_tasks_state_sync(flow_run_id, new_state):
    with get_client(sync_client=True) as client:
        task_runs = client.read_task_runs(
            task_run_filter=TaskRunFilter(
                flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id])
            )
        )
        results = []
        for tr in task_runs:
            if tr.state_type in {"COMPLETED", "FAILED", "CANCELLED", "CRASHED"}:
                continue
            res = client.set_task_run_state(tr.id, new_state)
            results.append((tr.id, res.accepted))
        return results

# set_all_tasks_state_sync(flow_run_id_y, Cancelled(message="..."))

Notes and caveats: - set_task_run_state returns an orchestration result with `accepted`—check it; invalid transitions may be rejected unless you pass

force=True

. Use force sparingly. - Cancelling running tasks: Prefect will attempt to cancel them; success depends on the worker/infrastructure. - This targets only tasks in flow run Y. If you also want to affect tasks in subflows spawned by Y, first find child flow runs (filter by parent_flow_run_id) and repeat. - Setting states to “Completed”/“Failed” for tasks still running can confuse orchestration; prefer “Cancelled” or let orchestration finish naturally unless you have a clear need.

Short answer - Setting a task run to CANCELLED is an orchestration signal; it does not forcibly stop Python code that’s already running. - If the task keeps running and later raises an exception, the server will typically keep the task’s state as CANCELLED. The task’s attempt to report FAILED will be rejected (unless someone overwrites with force=True). Why you see this - Orchestration and execution are separate. CANCELLED marks the run terminal in the API, but it cannot preemptively kill Python in-process work. - If you cancel before the task starts, it should not start. If you cancel after the task has already begun executing, it will usually continue running unless your infrastructure/worker can terminate it. What happens to the state after cancellation + exception - Subsequent state proposals (e.g., FAILED when the task ends with an exception) are evaluated by orchestration. From a terminal state (CANCELLED) they are generally rejected, so the task remains CANCELLED. - You can confirm by checking the result of a state set: - OrchestrationResult.accepted will be False if the server rejected the post-cancellation update. How to actually stop running work - Infrastructure-backed runs (Kubernetes/ECS/Docker workers): cancelling the run should prompt the worker to delete the job/pod/container; that actually ends the process. Ensure your worker type supports cancellation and has permissions to terminate the workload. - Local/process/thread workers: Python cannot be preempted; you must design cooperative cancellation. Options: - Periodically poll your own state and exit early when CANCELLED:

from prefect import task
    from prefect.context import get_run_context
    from prefect.client.orchestration import get_client
    import time

    @task
    def long_task():
        task_run_id = get_run_context().task_run.id
        with get_client(sync_client=True) as client:
            for chunk in range(1000):
                # do work...
                if chunk % 10 == 0:
                    tr = client.read_task_run(task_run_id)
                    if tr.state_type and str(tr.state_type) == "CANCELLED":
                        return  # or raise to stop early
                time.sleep(0.5)

- Or run the task in a subprocess/container you can kill on cancellation. If you share your worker type (process, Docker, Kubernetes, ECS, task workers), I can give exact steps to make cancellation actually stop the code (and how to confirm state transitions).

You need a valid Fernet key (base64-url-encoded 32 bytes). Here are easy, safe ways to generate one and set it for local runs. Fastest (no extra packages) - One-liner using Python stdlib:

python -c "import os, base64; print(base64.urlsafe_b64encode(os.urandom(32)).decode())"

Using cryptography (if installed)

python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

Validate a key (optional)

python - <<'PY'
from cryptography.fernet import Fernet
key = input("Paste key: ").strip()
Fernet(key.encode())
print("Key is valid.")
PY

Set the env var - Unix/macOS:

export PREFECT_SERVER_ENCRYPTION_KEY='<paste_generated_key>'

- Windows PowerShell:

setx PREFECT_SERVER_ENCRYPTION_KEY "<paste_generated_key>"

- Docker Compose (example):

services:
  prefect:
    image: prefecthq/prefect:3-latest
    environment:
      PREFECT_SERVER_ENCRYPTION_KEY: "<paste_generated_key>"

Important cautions - If you already have a Prefect DB with encrypted data, do not use a new random key. Extract the existing key from your DB’s configuration table (value->>'fernet_key') and set that exact value. Using a different key will make existing blocks unreadable. - All server instances (Orion) must use the same key. Agents/workers do not need it. If you tell me how you’re running Prefect locally (venv, Docker, Compose, etc.), I can give the exact snippet to inject the key there.

Likely causes for “cancelled but still runs” in prod while it stops locally: - Timing/race: If you cancel after tasks have already transitioned to RUNNING (or have been dispatched to infrastructure), they will continue executing. Locally you may be cancelling earlier (before tasks are scheduled). - Infra/worker behavior: - Process/local workers cannot preempt Python; tasks keep running unless your code cooperatively checks for cancellation and exits. - Infra-backed workers (Kubernetes/ECS/Docker) can actually stop pods/tasks/containers on cancel. If your prod worker lacks permissions or isn’t the right type, cancellation won’t kill the workload. - Different mechanism used: Setting state to Cancelled is not the same as issuing a cancel. Prefer the cancel action which triggers worker-side termination: - CLI:

prefect flow-run cancel <flow_run_id>

- API:

await client.set_flow_run_state(flow_run_id, Cancelled(...))

is allowed, but won’t kill already-running code; the cancel endpoint plus a cooperating worker is more reliable. - Rejection of your cancel attempt: State changes are not guaranteed. If orchestration rejects your update (e.g., a concurrent RUNNING transition), your code may think it cancelled but it didn’t. Always check the result: -

res = await client.set_task_run_state(tr.id, Cancelled(...))

-

if not res.accepted: ...

(log this — it’s common with races) What to check next 1) Verify your cancel operations are accepted:

res_fr = await client.set_flow_run_state(flow_run_id, Cancelled(message="…"))
print("flow cancel accepted:", res_fr.accepted)

trs = await client.read_task_runs(task_run_filter=TaskRunFilter(
    flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id])
))
for tr in trs:
    res = await client.set_task_run_state(tr.id, Cancelled(message="…"))
    print(tr.id, "accepted:", res.accepted)

2) Confirm worker type and capabilities in prod: - Process/local: won’t preempt; implement cooperative cancellation (periodically fetch your own state and exit) or run tasks in a killable subprocess. - Kubernetes/ECS/Docker workers: verify the worker has permissions to delete jobs/pods/tasks/containers on cancel and that you’re actually using that worker type. 3) Prefer the cancel action: - CLI:

prefect flow-run cancel <id>

- API: cancellation endpoints ensure the worker is signaled to terminate infra jobs. Setting terminal states directly doesn’t kill already-running work. 4) Avoid new work during cancel: - Pause the work pool to stop dispatch:

prefect work-pool pause <pool>

- Then cancel the flow run; this prevents new tasks from being scheduled while you’re cancelling. Why it “works locally” - You may be cancelling before tasks move to RUNNING, so nothing gets dispatched. - Your local engine might be synchronous with less concurrency, shrinking the race window. If you share: - Prefect version (prod vs local) - Worker type in prod (process/docker/kubernetes/ecs) and how flows/tasks are launched - How you trigger cancellation (CLI vs set_state) and whether you’re checking

accepted

I can give concrete steps to make cancellation reliably stop work in your setup.

What you’re seeing is consistent with how cancellation works: - Locally: you cancelled before tasks were submitted, so the engine never started them. Your “check state, log, and exit” code inside the task never ran because the task never started. That’s expected. - In prod: you’re likely cancelling after the flow worker has already started the flow and submitted (or begun) the tasks. With a Process worker, Prefect cannot preempt Python code; already-started tasks will keep running unless your task code cooperatively checks for cancellation and exits. Why your “check once at start” didn’t log in prod - If your task checks its state only once at the top, and the cancel arrives just after the start, that initial check will see RUNNING, not CANCELLED. The task then proceeds to sleep/work and you won’t see the “cancelled” log. Later, when you set CANCELLED, the task keeps running because it doesn’t check again. What to do 1) Check acceptance and ordering - Always inspect the orchestration result when you set states:

res = await client.set_task_run_state(task_run_id, Cancelled(...))
print("accepted:", res.accepted)

- If it’s False, your cancellation was rejected (e.g., a race with a RUNNING transition). - Inspect state history to confirm ordering:

states = await client.read_task_run_states(task_run_id)
for s in states: print(s.timestamp, s.name)

If RUNNING precedes CANCELLED, your cancel came too late to stop execution. 2) Make tasks cooperatively cancellable - Poll for cancellation during long work; don’t just check once:

from prefect import task
from prefect.context import get_run_context
from prefect.client.orchestration import get_client
from prefect.logging import get_run_logger
import time

@task
def long_task():
    log = get_run_logger()
    tid = get_run_context().task_run.id
    with get_client(sync_client=True) as client:
        for i in range(60):  # 60 x 1s units of work
            # do a small unit of work here
            if i % 5 == 0:  # check periodically
                tr = client.read_task_run(tid)
                if str(tr.state_type) == "CANCELLED":
                    <http://log.info|log.info>("Detected cancellation; exiting early")
                    return
            time.sleep(1)

3) Prevent new work while cancelling - Pause the work pool before cancelling so the worker doesn’t submit more tasks during your cancel window:

prefect work-pool pause <pool>
# then cancel the flow run
prefect flow-run cancel <flow_run_id>

4) Prefer the cancel action for flows - Use the cancel endpoint/CLI (

prefect flow-run cancel <id>

) to signal the worker; setting states directly is allowed, but won’t kill already-running code. With a Process worker, only cooperative checks will stop the task. Why local vs prod differs - Locally, you likely hit the cancel earlier (before task submission), so tasks never start—no “cancelled” log. - In prod, cancellation often races with scheduling; by the time you set CANCELLED, tasks have already started. Without periodic checks, they won’t stop. If you share: - How you trigger cancellation (CLI vs API) and whether

accepted

was True - A snippet of your task’s check logic - The state history ordering (Running vs Cancelled timestamps) I can help pinpoint the exact race and suggest where to add checks or adjust your cancel timing.

What you’re seeing comes down to timing and how cancellation works with a Process worker: - Setting states (flow run and task runs) is orchestration-only. It marks them Cancelled in the API but does not kill Python that’s already running. - With a Process worker, the flow code runs in a subprocess. If you cancel after the flow is Running and a task has already been accepted to run, that subprocess will keep executing unless the worker detects the cancellation and terminates the subprocess, or your task cooperatively exits. Why local vs prod looks different - Locally, you likely set Cancelled before the engine actually invoked the next task, so the server rejected the task’s transition to RUNNING and the task never started. That’s why your “cancelled” log inside the task never appeared: the task didn’t run at all. - In prod, you’re cancelling after the flow is already Running and tasks have been submitted. If you only set states (without the worker actually terminating the process), tasks that were already accepted to run will execute; later state updates to Cancelled will be retained, and the task’s attempt to report Completed/Failed will be rejected. What to do to reliably stop execution 1) Use the cancel action (don’t just set states): - CLI:

prefect flow-run cancel <flow_run_id>

- Or API:

async with get_client() as client:
      res = await client.set_flow_run_state(flow_run_id, Cancelled("..."))
      print("accepted:", res.accepted)

Then watch your Process worker logs; it should detect the cancellation and terminate the flow subprocess. If you don’t see log lines like “Cancelling flow run …” or “Terminating process …”, the worker isn’t acting on the cancel (bug or config issue). 2) Make long tasks cooperatively cancellable: - Don’t just check at the start; periodically poll your own state and exit early if Cancelled. This prevents long sleeps/work from continuing after cancel. 3) Order of operations to minimize races: - Cancel the flow run first (via cancel API/CLI). - Do not manually set task run states; let orchestration stop scheduling and reject further RUNNING transitions. - Confirm the orchestration result is accepted (res.accepted == True) and check state history ordering (Cancel before next RUNNING). 4) Debug checklist: - Confirm your cancel call was accepted (res.accepted True). - Check state history timestamps: Cancelled should be the latest before the next task would start. - Inspect Process worker logs for cancellation handling; you should see it terminating the child process. - Ensure both worker and server point to the same API URL/workspace. - Verify Prefect version is the same locally and in prod. Why your “check once” didn’t log - If you cancel after the task started, the initial check will see RUNNING, not CANCELLED. Without periodic checks, the task keeps going (sleeping for 60s) and you won’t see the “cancelled” log. If you can share: - How you trigger cancellation (CLI vs API) and whether accepted=True - Worker logs around the cancel time - Prefect version and how the Process worker is started I can help pinpoint whether it’s just timing/cooperative-cancel gaps or a worker not reacting to cancellation.

What you’re running into is a timing/orchestration gap with a Process worker: - Setting states via the API is advisory. It will not interrupt or “rewind” a Python process that’s already executing your flow code. - When you cancel during B, the task runs for C and D usually do not exist yet. So “set all task runs to Cancelled” only touches A/B; C/D will be created later when your flow code proceeds and calls C() and D(). Those new task runs are not pre-cancelled, so they execute. Why you see “C and D still run” - The flow function is still running in a subprocess. After B returns, the Python code simply calls C(), then D(). The server being Cancelled does not stop the Python function unless the worker terminates the subprocess or your code cooperatively stops. - Locally, you likely hit the cancel earlier (before the next task was created), so you see no further execution. In prod, by the time you cancel, the flow is already in the part of the code path that will call the next tasks. How to make cancellation actually stop C and D 1) Prefer the cancel action (don’t just set states) - CLI:

prefect flow-run cancel <flow_run_id>

- API:

from prefect.client.orchestration import get_client
from prefect.states import Cancelled

async with get_client() as client:
    res = await client.set_flow_run_state(flow_run_id, Cancelled("…"))
    print("accepted:", res.accepted)

With a Process worker, this should prompt the worker to terminate the flow subprocess so the flow cannot schedule C/D. Check worker logs for cancellation handling. 2) Add cooperative checks between tasks in the flow - Because a Process worker can’t preempt Python, add a small guard in your flow between task calls to bail early if the flow has been cancelled.

from prefect import flow
from prefect.context import get_run_context
from prefect.client.orchestration import get_client

def flow_cancelled_sync() -> bool:
    with get_client(sync_client=True) as client:
        fr = client.read_flow_run(get_run_context().flow_run.id)
        return str(fr.state_type) == "CANCELLED"

@flow
def my_flow():
    A()
    if flow_cancelled_sync(): return
    B()
    if flow_cancelled_sync(): return
    C()
    if flow_cancelled_sync(): return
    D()

This prevents new task runs from being created after cancellation. 3) Make long-running tasks poll for cancellation - Don’t just check once at the start; poll periodically and exit early. This stops in-flight work when cancellation arrives. 4) Always check that your cancel was accepted - Race conditions can cause a Cancelled proposal to be rejected. Inspect the orchestration result:

res = await client.set_task_run_state(tr.id, Cancelled("…"))
if not res.accepted:
    # log/handle — this means the server did not apply your cancel

5) Don’t rely on pre-cancelling future tasks - If C and D don’t exist yet, setting their state does nothing. Either cancel the flow so it can’t schedule them (preferred), or add the flow-level guard above. If you share a snippet of how you’re issuing the cancellations and a few lines from the Process worker logs around that time, I can help confirm whether the worker is receiving and acting on the cancel signal.

Thanks for the detailed scenario — given your description, there are only a few plausible reasons C and D still start after you cancel during B. What should happen - If the flow run is set to Cancelled before the engine attempts to create/schedule the next task (C), the server should reject new task state proposals and C/D should not start. - With a Process worker, if you use the cancel action for the flow run, the worker should terminate the flow subprocess so the Python code cannot proceed to C/D. Why prod may differ from your local run 1) Version mismatch or behavior change - If your local env is on a newer Prefect than prod, it may have a fix that prevents post-cancel scheduling. Verify Prefect versions for: - Server (Orion) in prod vs local - Process worker image/package in prod vs local 2) You’re setting states, not “cancelling” the flow (and racing) - Manually setting states to Cancelled can race with the engine’s proposals. - If the Cancelled proposal for the flow run is not accepted before C is created, the server can still accept C’s RUNNING proposal. - Always check orchestration results:

res = await client.set_flow_run_state(flow_run_id, Cancelled("..."))
  print("flow cancel accepted:", res.accepted)

If it’s False (or arrives too late), the engine may proceed. 3) You cancelled existing task runs only - When you enumerate and cancel “all task runs,” at that instant C and D do not exist. If the flow continues, it will create new task run records for C/D that were never pre-cancelled. Cancelling the flow run is the reliable gate. 4) Worker didn’t terminate the flow subprocess - With a Process worker, the cancel action should trigger termination of the child process. If the worker doesn’t react, the Python code will keep running and schedule C/D. - Check worker logs at cancel time for lines indicating it received the cancel and terminated the process. If you don’t see that, the worker may not be handling cancel events (bug/config). How to diagnose quickly - Confirm cancel acceptance ordering:

# 1) cancel flow run
  fr_res = await client.set_flow_run_state(flow_run_id, Cancelled("..."))
  print("flow cancel accepted:", fr_res.accepted)

  # 2) after B finishes, see what happened to C
  trs = await client.read_task_runs(task_run_filter=TaskRunFilter(
      flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id])
  ))
  # Look at state history and timestamps for C:
  states = await client.read_task_run_states(next(tr.id for tr in trs if tr.name=="C"))
  for s in states: print(s.timestamp, s.name)

You want to see Cancelled on the flow run before any state proposals for C. - Check worker logs around the cancel moment: - Do you see the worker acknowledging the flow cancel and terminating the subprocess? - If not, that’s likely the cause. Recommended changes - Prefer the cancel action for the flow run (CLI or API) instead of manually setting states:

prefect flow-run cancel <flow_run_id>

or (API) set Cancelled and verify

accepted=True

. - Add a small cooperative guard in the flow between tasks to prevent new tasks from being created if a race occurs:

from prefect.context import get_run_context
  from prefect.client.orchestration import get_client

  def flow_cancelled_sync() -> bool:
      with get_client(sync_client=True) as client:
          fr = client.read_flow_run(get_run_context().flow_run.id)
          return str(fr.state_type) == "CANCELLED"

  @flow
  def f():
      A()
      if flow_cancelled_sync(): return
      B()
      if flow_cancelled_sync(): return
      C()
      if flow_cancelled_sync(): return
      D()

This is a belt-and-suspenders approach for Process workers, which cannot preempt Python mid-function. - Ensure prod and local Prefect versions match for server and worker. If after these checks you still see C/D starting even when the flow shows Cancelled before they’re created, that’s unexpected — please share:

- Prefect versions (server and worker) local vs prod - The orchestration result (accepted flag) for the flow cancel - A few lines of worker logs at cancel time I can help you pin this down or file a bug if it’s reproducible.

ObjectNotFound from read_task_run at the very start of a task almost always means the client you’re calling is not looking at the same API/workspace as the run, or you’ve hit a tiny timing window before the server commit is visible. Here’s how to narrow it down and fix. Most likely causes - API/workspace mismatch - The task process is pointed at a different API than the one that created the run (different PREFECT_API_URL for OSS; or different Cloud account/workspace/env vars). - In Cloud, missing/incorrect PREFECT_API_KEY or ACCOUNT/WORKSPACE IDs in the task process will make read_task_run return 404 even though you’re inside a real run. - Tiny race at task start - You’re reading immediately at context entry and the DB row isn’t visible yet (commit/transaction timing). A short retry usually resolves this. Quick diagnostics (run inside the task)

from prefect import get_run_logger, __version__ as prefect_version
from prefect.settings import PREFECT_API_URL, PREFECT_CLOUD_API_URL, PREFECT_CLOUD_ACCOUNT_ID, PREFECT_CLOUD_WORKSPACE_ID

log = get_run_logger()
<http://log.info|log.info>(f"prefect_version={prefect_version}")
<http://log.info|log.info>(f"PREFECT_API_URL={PREFECT_API_URL.value()}")
# If using Cloud:
<http://log.info|log.info>(f"PREFECT_CLOUD_API_URL={PREFECT_CLOUD_API_URL.value()}")
<http://log.info|log.info>(f"ACCOUNT_ID={PREFECT_CLOUD_ACCOUNT_ID.value()} WORKSPACE_ID={PREFECT_CLOUD_WORKSPACE_ID.value()}")

Compare those to the values your worker/server use when the flow was created. Add a short retry to dodge the startup race

import asyncio
from prefect.context import get_run_context
from prefect.client.orchestration import get_client
from prefect.exceptions import ObjectNotFound

@task
async def t():
    tid = get_run_context().task_run.id
    async with get_client() as client:
        for _ in range(5):
            try:
                tr = await client.read_task_run(tid)
                return tr
            except ObjectNotFound:
                await asyncio.sleep(0.2)
        raise

Other checks - Verify you’re calling the async client from an async task and the sync client from a sync task (mismatching won’t cause ObjectNotFound, but good to confirm). - Make sure you didn’t accidentally use a different run id (e.g., flow_run.id) when calling read_task_run. If the retry still 404s and your env values match, try reading the flow run first:

fr = await client.read_flow_run(get_run_context().flow_run.id)

If that also 404s, it’s almost certainly an API/workspace mismatch. If flow run works but task run doesn’t, it’s a timing issue — increase the retry window slightly (e.g., 1–2 seconds) and it should resolve. If you share the logged API/workspace values from inside the task and on the server/worker, I can help pinpoint the mismatch.