api/task_runs/{id}/set_state

on_completion

on_failure

on_crashed

set_state

python
from prefect import flow
from prefect.client import get_client

async def cancel_and_cleanup(flow_run_id: str):
    async with get_client() as client:
        # First, cancel the flow run
        await client.set_flow_run_state(
            flow_run_id=flow_run_id,
            state={"type": "CANCELLED"}
        )
        
        # Then trigger your cleanup flow
        await cleanup_flow(flow_run_id=flow_run_id)

@flow
async def cleanup_flow(flow_run_id: str):
    # Your cleanup logic here
    ...

allow_failure(C)

allow_failure(C)

python
from prefect import flow, task, allow_failure

@task
async def task_a():
    return "A"

@task
async def task_b():
    return "B"

@task
async def task_c():
    return "C"

@task
async def task_d():
    return "D"

@flow
async def my_flow():
    a = await task_a()
    b = await task_b(wait_for=[a])
    c = await task_c(wait_for=[b])
    # D will run even if C is cancelled because of allow_failure
    d = await task_d(wait_for=[allow_failure(c)])

allow_failure(C)

allow_failure

allow_failure(C)

allow_failure

python
@flow
async def my_flow():
    a = await task_a()
    b = await task_b(wait_for=[a])
    c = await task_c(wait_for=[b])
    # D runs independently
    d = await task_d()

python
@flow
async def my_flow():
    # Start both paths simultaneously
    results = await anyio.create_task_group(
        task_a(),  # This starts the A -> B -> C chain
        task_d()   # This runs independently
    )

Block.load()

python
from prefect.blocks.core import Block

# Load a block document by name
my_block = Block.load("your-block-name")

python
from prefect.client import get_client

async def get_block_document(block_document_id):
    client = get_client()
    block_document = await client.read_block_document(
        block_document_id,
        include_secrets=True  # Set to False if you don't need secret values
    )
    return block_document

Block.load()

PrefectClient

Block.load()

python
from prefect.blocks.core import Block

# Load a block by name
block = Block.load("your-block-name")

python
from prefect.client import get_client

async def get_block_document(block_document_id):
    client = get_client()
    block_document = await client.read_block_document(
        block_document_id,
        include_secrets=True
    )
    return block_document

block_document

data

data

json
{
    "block_type_slug": "slack-webhook",
    "configuration": {
        "url": "<https://hooks.slack.com/...>",
        "timeout": 60
        // ... other block-specific configuration
    }
}

configuration

configuration

data

block_type_id

block_schema_id

sql
SELECT data->>'block_type_slug' as block_type,
       data->'configuration' as config
FROM block_document
WHERE id = 'your-block-id';

anonymous:<hash>

anonymous-<uuid>

"anonymous:48388c349c907ec7959e911b9d42eebd"

is_anonymous

sql
-- First, backup your existing named blocks (non-anonymous ones)
CREATE TEMP TABLE backup_blocks AS 
SELECT * FROM block_document 
WHERE is_anonymous = false;

-- Delete only anonymous blocks that are older than a certain period
-- Adjust the interval as needed (e.g., '7 days'::interval for weekly cleanup)
DELETE FROM block_document 
WHERE is_anonymous = true 
AND created < (NOW() - '30 days'::interval);

-- If something goes wrong, you can restore your named blocks
-- INSERT INTO block_document SELECT * FROM backup_blocks;

python
from prefect import flow
from prefect.blocks.system import JSON
import sqlalchemy as sa

@flow
def cleanup_anonymous_blocks():
    # Using SQLAlchemy for safety
    engine = sa.create_engine('your_database_url')
    with engine.begin() as conn:
        conn.execute(sa.text("""
            DELETE FROM block_document 
            WHERE is_anonymous = true 
            AND created < (NOW() - interval '30 days')
        """))

sql
-- See what's creating anonymous blocks
SELECT 
    substring(name from 1 for 50) as block_name,
    created,
    block_type_name,
    is_anonymous
FROM block_document
WHERE is_anonymous = true
ORDER BY created DESC
LIMIT 10;

sql
-- Safe cleanup of old anonymous blocks
DELETE FROM block_document
WHERE is_anonymous = true
AND created < (CURRENT_TIMESTAMP - interval '7 days')
AND name NOT LIKE 'prefect-system-%'; -- Protect any system-related blocks

sql
-- Monitor block growth over time
SELECT 
    date_trunc('day', created) as day,
    count(*) as total_blocks,
    sum(case when is_anonymous then 1 else 0 end) as anonymous_blocks
FROM block_document
GROUP BY date_trunc('day', created)
ORDER BY day DESC;