my_flow()

from threading import Thread
  from prefect import flow

  @flow
  def my_flow(n: int):
      ...

  t1 = Thread(target=my_flow, args=(1,))
  t2 = Thread(target=my_flow, args=(2,))
  t1.start(); t2.start()
  t1.join(); t2.join()

import asyncio
  from prefect import flow

  @flow
  async def my_flow(n: int):
      ...

  asyncio.run(asyncio.gather(my_flow(1), my_flow(2)))

ThreadPoolTaskRunner

ProcessPoolTaskRunner

@flow

async def

await

asyncio.to_thread

def

asyncio.to_thread(...)

run_sync_in_worker_thread(...)

from prefect import flow

@flow
async def process(data):
    ...

async def handle_request(req):
    result = await process(req.data)  # no to_thread needed
    return result

import asyncio
from prefect import flow

@flow
def process(data):
    ...

async def handle_request(req):
    result = await asyncio.to_thread(process, req.data)
    return result

from prefect import flow
from prefect.utilities.asyncutils import run_sync_in_worker_thread

@flow
def process(data):
    ...

async def handle_request(req):
    result = await run_sync_in_worker_thread(process, req.data)
    return result

from prefect import flow, task

@task(cache_result_in_memory=False, persist_result=False)
def step(...):
    ...

@flow(cache_result_in_memory=False, persist_result=False, log_prints=False)
def my_flow(...):
    ...

persist_result=True

f = some_task.submit(...)
result = f.result()
del f, result

from prefect import flow
from prefect.task_runners import ProcessPoolTaskRunner

@flow(task_runner=ProcessPoolTaskRunner(), cache_result_in_memory=False)
def my_flow(...):
    ...

from prefect.flow_engine import run_flow_in_subprocess

proc = run_flow_in_subprocess(my_flow, parameters={"x": 1})
proc.join()
# exchange results via files/object storage if needed

@flow(log_prints=False)

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task(cache_result_in_memory=False, persist_result=False)
def io_step(...):
    ...

@flow(
    cache_result_in_memory=False,
    persist_result=False,
    log_prints=False,  # avoid log buffering
    task_runner=ThreadPoolTaskRunner(max_workers=8)  # good for I/O-bound
)
def my_flow(...):
    # chain steps without retaining large intermediates
    a = io_step(...)
    b = io_step(a)
    ...
    return final_small_result

f = io_step.submit(...)
res = f.result()
# use res then
del f, res

ThreadPoolTaskRunner

max_workers

log_prints=False

from prefect.flow_engine import run_flow_in_subprocess

# Parent service code:
proc = run_flow_in_subprocess(my_flow, parameters={"...": "..."})
proc.join()
# For large final results, write to a file/object store inside the flow and read back here.

from prefect.utilities.asyncutils import run_sync_in_worker_thread

async def handle_request(...):
    result = await run_sync_in_worker_thread(my_flow, ...)
    return result

async def

await my_flow(...)

run_flow_in_subprocess

import json
from prefect import flow, task

@task(cache_result_in_memory=False, persist_result=False)
def io_step(val: int) -> dict:
    # do I/O-bound work and produce a small result
    return {"value": val}

@flow(cache_result_in_memory=False, persist_result=False, log_prints=False)
def my_flow(x: int, output_path: str) -> None:
    # chain your ~20 tasks here; only final result matters
    final = io_step(x)
    with open(output_path, "w") as f:
        json.dump(final, f)

import os
import json
import tempfile
from prefect.flow_engine import run_flow_in_subprocess

def run_my_flow_and_get_result(x: int) -> dict:
    with tempfile.TemporaryDirectory() as tmpdir:
        output_path = os.path.join(tmpdir, "result.json")

        proc = run_flow_in_subprocess(my_flow, parameters={"x": x, "output_path": output_path})
        proc.join()

        if proc.exitcode != 0:
            raise RuntimeError(f"Flow subprocess failed with exit code {proc.exitcode}")

        with open(output_path, "r") as f:
            return json.load(f)

import asyncio
from prefect.utilities.asyncutils import run_sync_in_worker_thread

async def run_my_flow_and_get_result_async(x: int) -> dict:
    return await run_sync_in_worker_thread(run_my_flow_and_get_result, x)

cache_result_in_memory=False

persist_result=False

proc.exitcode

@flow(persist_result=True, result_storage="/path", result_serializer=...)

output_path

# flows.py
from prefect import flow, task

@task(cache_result_in_memory=False, persist_result=False)
def step(x: int) -> dict:
    # I/O-bound work here
    return {"value": x}

@flow(cache_result_in_memory=False, persist_result=False, log_prints=False)
def my_flow(x: int) -> dict:
    # ~20 tasks chained; only final result matters
    return step(x)

# service.py
import multiprocessing as mp
from flows import my_flow

def _run_flow_worker(x: int, result_queue: "mp.Queue"):
    try:
        result = my_flow(x)
        result_queue.put(("ok", result))
    except Exception as e:
        # send error info back to parent
        result_queue.put(("err", repr(e)))

def run_my_flow_in_process(x: int):
    ctx = mp.get_context("spawn")  # cross-platform-safe
    q = ctx.Queue()
    p = ctx.Process(target=_run_flow_worker, args=(x, q))
    p.start()
    p.join()

    if p.exitcode != 0:
        # child crashed; no safe result to read
        raise RuntimeError(f"Flow subprocess exited with code {p.exitcode}")

    status, payload = q.get_nowait()
    if status == "err":
        raise RuntimeError(f"Flow failed: {payload}")
    return payload

if __name__ == "__main__":
    print(run_my_flow_in_process(123))

# inside _run_flow_worker
import json, os, tempfile
...
result = my_flow(x)
with open(output_path, "w") as f:
    json.dump(result, f)
result_queue.put(("ok", output_path))

from prefect.utilities.asyncutils import run_sync_in_worker_thread

async def run_my_flow_async(x: int):
    return await run_sync_in_worker_thread(run_my_flow_in_process, x)

run_sync_in_worker_thread

from prefect.utilities.asyncutils import run_sync_in_worker_thread

await

await run_sync_in_worker_thread(__fn, *args, **kwargs)

*args

__fn

cancellable

limiter

limiter

get_thread_limiter()

CapacityLimiter

run_sync_in_worker_thread

cancellable

limiter

functools.partial

from prefect import flow
from prefect.utilities.asyncutils import run_sync_in_worker_thread
import requests

def fetch(url, timeout=10):
    return requests.get(url, timeout=timeout).json()

@flow
async def my_flow():
    data = await run_sync_in_worker_thread(fetch, "<https://httpbin.org/get>", timeout=10)
    return data

if __name__ == "__main__":
    my_flow()

from prefect import task, flow
from prefect.utilities.asyncutils import run_sync_in_worker_thread
import time

def slow_blocking(n):
    time.sleep(n)
    return f"slept {n}s"

@task
async def do_work(n: int):
    return await run_sync_in_worker_thread(slow_blocking, n)

@flow
async def my_flow():
    res = await do_work(2)
    print(res)

if __name__ == "__main__":
    my_flow()

from prefect import flow, task
from prefect.utilities.asyncutils import run_sync_in_worker_thread
from anyio import CapacityLimiter
import time

def io_heavy(x):
    time.sleep(1)
    return x

limiter = CapacityLimiter(5)  # at most 5 threads for this workload

@task
async def work_item(x):
    return await run_sync_in_worker_thread(io_heavy, x, limiter=limiter)

@flow
async def my_flow():
    results = await work_item.map(range(20))  # fan-out but with a per-call limiter
    return results

from functools import partial
from prefect.utilities.asyncutils import run_sync_in_worker_thread

def fn_with_cancellable_kw(arg, cancellable=False):
    # your function that happens to use 'cancellable'
    ...

partial_fn = partial(fn_with_cancellable_kw, cancellable=True)
result = await run_sync_in_worker_thread(partial_fn, "value")

run_sync_in_worker_thread

task1.map(...)

task2.submit(...)

task2

task1

task2

task2

task2

task1

task2

x1 = task1.map(x0)
  x2 = task2.map(x1)  # each task2[i] waits only for task1[i]

wait_for

x1 = task1.map(x0)
  x2 = task2.submit(arg, wait_for=x1)  # waits for all x1 to finish, doesn’t consume their values

x1_states = task1.map(x0, return_state=True)
  x2 = task2.submit(x1_states)  # task2 can inspect which items succeeded/failed

x1 = task1.map(x0)
  partial = x1.result(raise_on_failure=False)  # list of results or exceptions
  x2 = task2.submit(partial)

resolve_futures_to_results

resolve_futures_to_states

PrefectFutureList.result

from prefect import task as prefect_task
from prefect import get_run_logger
from prefect.cache_policies import NO_CACHE

def task(*args, **kwargs):
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)
    kwargs.setdefault("log_prints", True)  # 捕获 print
    return prefect_task(*args, **kwargs)

@task()  # 你的装饰器
def add(x, y):
    print(f"[add] inputs: x={x}, y={y}")
    result = x + y
    print(f"[add] output: {result}")
    return result

on_completion

task_run.parameters

get_run_logger()

from prefect import task as prefect_task, get_run_logger
from prefect.cache_policies import NO_CACHE

# 完成后记录输入与输出
async def _log_on_completion(task, task_run, state):
    logger = get_run_logger()
    try:
        <http://logger.info|logger.info>(f"[{task.name}] inputs: {task_run.parameters}")
    except Exception as e:
        logger.warning(f"[{task.name}] cannot read parameters: {e}")
    try:
        result = await state.result()  # 获取任务返回值
        <http://logger.info|logger.info>(f"[{task.name}] output: {result!r}")
    except Exception as e:
        logger.warning(f"[{task.name}] cannot resolve result: {e}")

# 可选：失败时也记录（包含异常）
async def _log_on_failure(task, task_run, state):
    logger = get_run_logger()
    try:
        exc = await state.result(raise_on_failure=False)
        logger.error(f"[{task.name}] failed: {exc!r}; inputs: {task_run.parameters}")
    except Exception as e:
        logger.error(f"[{task.name}] failed and could not read details: {e}")

def task(*args, **kwargs):
    """
    你的自定义 task 装饰器：统一默认参数，并自动记录输入输出
    """
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)
    kwargs.setdefault("log_prints", True)  # 如果你也想捕获 print

    # 合并用户已传入的 hooks
    on_completion = list(kwargs.pop("on_completion", []) or [])
    on_completion.append(_log_on_completion)
    kwargs["on_completion"] = on_completion

    on_failure = list(kwargs.pop("on_failure", []) or [])
    on_failure.append(_log_on_failure)
    kwargs["on_failure"] = on_failure

    return prefect_task(*args, **kwargs)

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

repr(result)

.submit()

aigc_video_res_list

.result()

preproc_video_effect

blades_req

preproc_merge

preproc_merge

video_effect_predata = preproc_video_effect.submit(
    logid, planning_tts_json, tts_res, video_path_list_url, card_res_list, ttv_res_list, exp_id
)

# True 表示有 AIGC 结果，保留原音频；False 则需要调用 blades 做特效
keep_ori_audio = bool(aigc_video_res_list)

video_effect_res = (
    video_effect_predata
    if keep_ori_audio
    else blades_req.submit(video_effect_predata)  # 传 Future 没问题，Prefect 会自动解析
)

merge_predata = preproc_merge.submit(video_effect_res, voice_effect_res, keep_ori_audio)
merge_res = blades_req.submit(merge_predata)

aigc_video_res_list

keep_ori_audio = bool(aigc_video_res_list.result())

@task
  def has_aigc(xs): return bool(xs)

  keep_ori_audio = has_aigc.submit(aigc_video_res_list)
  # 这里 keep_ori_audio 会是 Future，作为参数传给 preproc_merge 也会被自动解析

preproc_merge

video_effect_res

preproc_merge

@task
  def normalize_effect_data(data): 
      # 归一化到 preproc_merge 需要的 schema
      return normalized

  normalized_effect = normalize_effect_data.submit(video_effect_res)
  merge_predata = preproc_merge.submit(normalized_effect, voice_effect_res, keep_ori_audio)

preproc_video_effect

blades_req

preproc_merge

prefect

PREFECT_API_URL

/api

http://<host>:4200/api

curl -i http://<host>:<port>/api/health
  curl -i http://<host>:<port>/api/readiness
  curl -sS http://<host>:<port>/api/admin/version

/api/health

PREFECT_LOGGING_ROOT_LEVEL=DEBUG \
  PREFECT_SERVER_DATABASE_ECHO=true \
  prefect server start --log-level DEBUG

docker logs -f <prefect-server-container>

kubectl logs -f <pod> -c <container>

prefect server services start

# 分别在客户端和服务端环境执行
  prefect version

prefect config view --show-defaults | grep -i -E "DATABASE|API_URL|SERVER"

prefect server database upgrade -y

PREFECT_SERVER_DATABASE_TIMEOUT=30 prefect server database upgrade -y

pg_trgm

CREATE EXTENSION IF NOT EXISTS pg_trgm;

prefect server services list-services

PREFECT_SERVER_SERVICES_SCHEDULER_ENABLED=false

--no-services

detail

prefect version

PREFECT_API_URL

/api/health

/api/readiness

# 先找容器名
  docker ps --format 'table {{.Names}}\t{{.Image}}\t{{.Status}}\t{{.Ports}}'

  # 跟随输出
  docker logs -f <容器名>

  # 只看最近 200 行并跟随
  docker logs --tail=200 -f <容器名>

  # 只看最近 30 分钟
  docker logs --since=30m <容器名>

# 查看所有服务日志
  docker compose logs -f

  # 只看某个服务（例如 api / services）
  docker compose ps
  docker compose logs -f <service_name>

# stdout
  tail -F /proc/1/fd/1
  # stderr
  tail -F /proc/1/fd/2

  # 可用 ps 看下 PID 1 是什么
  ps aux

# docker run/compose 的 environment:
  PREFECT_LOGGING_ROOT_LEVEL=DEBUG
  PREFECT_SERVER_DATABASE_ECHO=true

  # 或在 command 里
  prefect server start --log-level DEBUG

docker ps

docker compose ps

from prefect import flow, task, get_run_logger
from prefect.futures import as_completed

@task
def boom(i: int) -> int:
    if i % 2:
        raise ValueError(f"bad {i}")
    return i * i

@flow
def main():
    logger = get_run_logger()
    futures = {i: boom.submit(i) for i in range(5)}

    # 按提交顺序处理
    for i, fut in futures.items():
        try:
            res = fut.result()  # 这里失败会抛异常
            <http://logger.info|logger.info>(f"boom({i}) -> {res}")
        except Exception as exc:
            logger.error(f"Task boom(i={i}) 失败, task_run_id={fut.task_run_id}, exc={exc}")

    # 或者按完成顺序处理
    for fut in as_completed(list(futures.values())):
        try:
            res = fut.result()
        except Exception as exc:
            # 通过反查 key 知道是哪个 i
            i = next(k for k, v in futures.items() if v is fut)
            logger.error(f"Task boom(i={i}) 失败, task_run_id={fut.task_run_id}, exc={exc}")

from prefect import flow, task, get_run_logger, allow_failure
from prefect.states import get_state_exception

@task
def boom(i: int) -> int:
    if i % 2:
        raise ValueError(f"bad {i}")
    return i * i

@flow
def main():
    logger = get_run_logger()
    futures = {i: allow_failure(boom.submit(i)) for i in range(5)}

    for i, fut in futures.items():
        st = fut.wait()   # 拿到 State，不会抛异常
        if st.is_failed():
            exc = get_state_exception(st)
            logger.error(f"boom({i}) 失败, task_run_id={fut.task_run_id}, exc={exc}")
        else:
            <http://logger.info|logger.info>(f"boom({i}) -> {st.result()}")

task_run_id

from prefect.client.orchestration import get_client

@flow
def main():
    fut = boom.submit(1)
    try:
        fut.result()
    except Exception:
        with get_client() as client:
            tr = client.read_task_run(fut.task_run_id)
            # tr.name, tr.task_key, tr.tags, tr.parameters 等都可用
            print(f"失败任务: name={tr.name}, id={tr.id}")

future.result()

future.task_run_id

Failed

raise

future.result()

from prefect import flow, task, get_run_logger, allow_failure
from prefect.futures import resolve_futures_to_states
from prefect.client.orchestration import get_client

@task
def boom(i: int):
    if i % 2:
        raise ValueError(f"bad {i}")
    return i*i

@flow
def main(n: int = 1000):
    logger = get_run_logger()

    # 1) 统一收集 future，并允许失败不阻塞后续（关键）
    futures = [allow_failure(boom.submit(i)) for i in range(n)]

    # 2) 一次性把所有 future 解析为 State
    states = resolve_futures_to_states(futures)

    # 3) 找到失败的 futures（不需要逐个 result()）
    failed = [f for f, st in states.items() if st.is_failed()]
    <http://logger.info|logger.info>(f"共失败 {len(failed)} 个 task")

    # 4) 想拿“任务名”，可用 task_run_id 反查
    if failed:
        from prefect.states import get_state_exception
        with get_client() as client:
            for f in failed:
                tr = client.read_task_run(f.task_run_id)
                exc = get_state_exception(states[f])
                logger.error(f"Task 失败: name={tr.name}, id={tr.id}, exc={exc}")

allow_failure(...)

resolve_futures_to_states

is_failed()

future.task_run_id

client.read_task_run(...)

from prefect import task, get_run_logger
from prefect.context import get_run_context

@task
def wrapped(...):
    logger = get_run_logger()
    ctx = get_run_context()   # 在 task 中可拿到 TaskRun 上下文
    try:
        return real_work(...)
    except Exception as exc:
        logger.error(f"Task 失败: name={ctx.task_run.name}, id={ctx.task_run.id}, exc={exc}")
        raise   # 仍然让任务真正失败

from prefect.futures import as_completed

futures = [boom.submit(i) for i in range(n)]
for fut in as_completed(futures):
    try:
        _ = fut.result()
    except Exception as exc:
        # 通过 fut.task_run_id 反查名字
        ...

prefect.states.Failed

task_run_id

P1D

-- 看前 20 大表（含索引）
SELECT relname AS table_name,
       pg_size_pretty(pg_total_relation_size(oid)) AS total_size
FROM pg_class
WHERE relkind = 'r'
ORDER BY pg_total_relation_size(oid) DESC
LIMIT 20;

-- 每张表的行数估计
SELECT relname, n_live_tup
FROM pg_stat_user_tables
ORDER BY n_live_tup DESC
LIMIT 20;

event

log

_state

PREFECT_SERVER_EVENTS_RETENTION_PERIOD

-- 预览将要删除的事件范围（举例：删除 7 天前）
SELECT COUNT(*) FROM event WHERE occurred < now() - interval '7 days';

-- 真正删除
DELETE FROM event WHERE occurred < now() - interval '7 days';

-- 预览
SELECT COUNT(*) FROM log WHERE "timestamp" < now() - interval '7 days';

-- 删除
DELETE FROM log WHERE "timestamp" < now() - interval '7 days';

-- 预览 30 天前完成的 flow run
SELECT COUNT(*) FROM flow_run
WHERE end_time < now() - interval '30 days';

-- 删除
DELETE FROM flow_run
WHERE end_time < now() - interval '30 days';

PrefectClient

VACUUM FULL

environment:
  # 强烈建议：减少或关闭日志上报
  PREFECT_LOGGING_TO_API_ENABLED: "false"
  PREFECT_LOGGING_LEVEL: "WARNING"

  # 缩短 events 保留期
  PREFECT_SERVER_EVENTS_RETENTION_PERIOD: "P1D"

2d

P2D

prefect config set PREFECT_SERVER_EVENTS_RETENTION_PERIOD="P2D"

PREFECT_EVENTS_RETENTION_PERIOD

prefect config validate
prefect config view | grep EVENTS_RETENTION_PERIOD

prefect config set

PREFECT_SERVER_EVENTS_RETENTION_PERIOD="P2D"

PREFECT_LOGGING_LEVEL=WARNING

-- 哪些表引用了某个表（例如 event_resources）：
SELECT
  tc.table_schema, tc.table_name, kcu.column_name,
  ccu.table_name AS foreign_table_name, ccu.column_name AS foreign_column_name,
  tc.constraint_name
FROM information_schema.table_constraints AS tc
JOIN information_schema.key_column_usage AS kcu
  ON tc.constraint_name = kcu.constraint_name
JOIN information_schema.constraint_column_usage AS ccu
  ON ccu.constraint_name = tc.constraint_name
WHERE tc.constraint_type = 'FOREIGN KEY'
  AND ccu.table_name = 'event_resources';

-- 预览
SELECT COUNT(*) FROM event WHERE occurred < now() - interval '3 days';

-- 分批删除（每批 100 万行，循环执行多次）
DELETE FROM event
WHERE occurred < now() - interval '3 days'
ORDER BY occurred
LIMIT 1000000;

-- 示例：如果 event 表通过 event.resource_id 指向 event_resources(id)
DELETE FROM event_resources er
WHERE NOT EXISTS (
  SELECT 1 FROM event e WHERE e.resource_id = er.id
);

-- 预览
SELECT COUNT(*) FROM log WHERE "timestamp" < now() - interval '7 days';

-- 分批删除
DELETE FROM log
WHERE "timestamp" < now() - interval '7 days'
ORDER BY "timestamp"
LIMIT 100000;

-- 先找出要删除的旧 flow_run
WITH old_flows AS (
  SELECT id FROM flow_run
  WHERE end_time IS NOT NULL
    AND end_time < now() - interval '30 days'
  LIMIT 50000  -- 分批
)

-- 预览计数
SELECT COUNT(*) FROM old_flows;

-- 先删与旧 flow 相关的日志（如果没在前面按时间统一清过）
DELETE FROM log
WHERE flow_run_id IN (SELECT id FROM old_flows);

-- 删 task_run_state（依赖 task_run）
DELETE FROM task_run_state
WHERE task_run_id IN (
  SELECT id FROM task_run
  WHERE flow_run_id IN (SELECT id FROM old_flows)
);

-- 删 task_run
DELETE FROM task_run
WHERE flow_run_id IN (SELECT id FROM old_flows);

-- 删 flow_run_state（依赖 flow_run）
DELETE FROM flow_run_state
WHERE flow_run_id IN (SELECT id FROM old_flows);

-- 删 flow_run
DELETE FROM flow_run
WHERE id IN (SELECT id FROM old_flows);

VACUUM (ANALYZE) event;
VACUUM (ANALYZE) event_resources;
VACUUM (ANALYZE) log;
VACUUM (ANALYZE) task_run_state;
VACUUM (ANALYZE) task_run;
VACUUM (ANALYZE) flow_run_state;
VACUUM (ANALYZE) flow_run;

pg_repack

import os
import time
from datetime import datetime, timedelta, timezone

from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterStartTime,
    FlowRunFilterState,
    FlowRunFilterStateType,
    ArtifactFilter,
    ArtifactFilterFlowRunId,
)
from prefect.client.schemas.objects import StateType

# 配置
# 请确保运行环境已设置 PREFECT_API_URL 指向你的自托管 Server，例如：
# os.environ["PREFECT_API_URL"] = "http://<your-server-host>:4200/api"
RETENTION_DAYS_FOR_RUNS = 30           # 仅保留近 30 天的 flow runs（示例）
BATCH_SIZE = 500                       # 每批处理条数（按环境调节）
SLEEP_BETWEEN_BATCHES_SEC = 0.2        # 批次间隔，避免打爆 API/DB
DRY_RUN = True                         # 先 dry-run 看看命中数量，再设为 False 执行删除

# 仅删除“已结束”的 runs，避免误删进行中的 runs
END_STATES = [
    StateType.COMPLETED,
    StateType.FAILED,
    StateType.CANCELLED,
    StateType.CRASHED,
]


def delete_artifacts_for_run(client: SyncPrefectClient, flow_run_id):
    # 按 flow_run_id 过滤 artifacts 并删除
    # 分页拉取，直到为空
    total_deleted = 0
    while True:
        arts = client.read_artifacts(
            artifact_filter=ArtifactFilter(
                flow_run_id=ArtifactFilterFlowRunId.any_([flow_run_id])
            ),
            limit=BATCH_SIZE,
            offset=0,  # 我们每次都从 0 拉，因为会不断删除，集合会缩小
        )
        if not arts:
            break
        for a in arts:
            if DRY_RUN:
                print(f"[DRY] would delete artifact {a.id} (key={getattr(a, 'key', None)}) for flow_run {flow_run_id}")
            else:
                client.delete_artifact(a.id)
            total_deleted += 1
        if SLEEP_BETWEEN_BATCHES_SEC:
            time.sleep(SLEEP_BETWEEN_BATCHES_SEC)
    return total_deleted


def main():
    cutoff = datetime.now(timezone.utc) - timedelta(days=RETENTION_DAYS_FOR_RUNS)

    with SyncPrefectClient() as client:
        print(f"Start cleanup. Retention for runs: {RETENTION_DAYS_FOR_RUNS} days; cutoff={cutoff.isoformat()}; DRY_RUN={DRY_RUN}")

        total_runs_deleted = 0
        total_artifacts_deleted = 0

        while True:
            # 每次取一批“早于 cutoff 且已结束”的 flow runs
            runs = client.read_flow_runs(
                flow_run_filter=FlowRunFilter(
                    start_time=FlowRunFilterStartTime(before_=cutoff),
                    state=FlowRunFilterState(
                        type=FlowRunFilterStateType.any_(END_STATES)
                    ),
                ),
                limit=BATCH_SIZE,
                offset=0,  # 结合删除操作，始终从 0 开始拿
            )
            if not runs:
                break

            print(f"Found {len(runs)} old flow runs to delete in this batch...")

            for r in runs:
                # 先删 artifacts（如果你从不使用 artifacts，也可以跳过这步）
                deleted_art_count = delete_artifacts_for_run(client, r.id)
                total_artifacts_deleted += deleted_art_count

                # 再删 flow run（一般会级联删除该 run 下的 task runs 与 states）
                if DRY_RUN:
                    print(f"[DRY] would delete flow run {r.id} (name={getattr(r, 'name', None)}, start={getattr(r, 'start_time', None)})")
                else:
                    client.delete_flow_run(r.id)
                total_runs_deleted += 1

            if SLEEP_BETWEEN_BATCHES_SEC:
                time.sleep(SLEEP_BETWEEN_BATCHES_SEC)

        print(f"Cleanup done. runs_deleted={total_runs_deleted}, artifacts_deleted={total_artifacts_deleted}, DRY_RUN={DRY_RUN}")


if __name__ == "__main__":
    main()

PREFECT_LOGGING_TO_API_ENABLED=false

PREFECT_LOGGING_LEVEL=WARNING/ERROR

-- 建议先确保有时间列索引，加速筛选
-- CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_ts ON log ("timestamp");

WITH to_delete AS (
  SELECT id
  FROM log
  WHERE "timestamp" < now() - interval '30 days'
  ORDER BY "timestamp"
  LIMIT 100000
)
DELETE FROM log l
USING to_delete d
WHERE l.id = d.id;

WITH to_delete AS (
  SELECT ctid
  FROM log
  WHERE "timestamp" < now() - interval '30 days'
  ORDER BY "timestamp"
  LIMIT 100000
)
DELETE FROM log l
USING to_delete d
WHERE l.ctid = d.ctid;

DO $$
DECLARE
  rows_deleted integer;
BEGIN
  LOOP
    WITH to_delete AS (
      SELECT ctid
      FROM log
      WHERE "timestamp" < now() - interval '30 days'
      LIMIT 100000
    )
    DELETE FROM log l
    USING to_delete d
    WHERE l.ctid = d.ctid;

    GET DIAGNOSTICS rows_deleted = ROW_COUNT;
    EXIT WHEN rows_deleted = 0;
    PERFORM pg_sleep(0.1); -- 批次间隙，避免长时间锁
  END LOOP;
END
$$;

WITH to_delete AS (
  SELECT ctid
  FROM event
  WHERE occurred < now() - interval '3 days'
  LIMIT 100000
)
DELETE FROM event e
USING to_delete d
WHERE e.ctid = d.ctid;

pg_repack

PREFECT_LOGGING_LEVEL=WARNING

import os
import time
from datetime import datetime, timedelta, timezone

from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterStartTime,
    FlowRunFilterState,
    FlowRunFilterStateType,
    ArtifactFilter,
    ArtifactFilterFlowRunId,
)
from prefect.client.schemas.objects import StateType

# 请确保运行环境有 PREFECT_API_URL 指向你的自托管 Server，如：
# os.environ["PREFECT_API_URL"] = "http://<server-host>:4200/api"

RETENTION_DAYS_FOR_RUNS = 15
BATCH_SIZE = 500
SLEEP_BETWEEN_BATCHES_SEC = 0.2
DRY_RUN = True  # 先 dry-run；确认无误后改为 False 执行删除

END_STATES = [
    StateType.COMPLETED,
    StateType.FAILED,
    StateType.CANCELLED,
    StateType.CRASHED,
]

def delete_artifacts_for_run(client: SyncPrefectClient, flow_run_id):
    total_deleted = 0
    while True:
        arts = client.read_artifacts(
            artifact_filter=ArtifactFilter(
                flow_run_id=ArtifactFilterFlowRunId.any_([flow_run_id])
            ),
            limit=BATCH_SIZE,
            offset=0,
        )
        if not arts:
            break
        for a in arts:
            if DRY_RUN:
                print(f"[DRY] would delete artifact {a.id} for flow_run {flow_run_id}")
            else:
                client.delete_artifact(a.id)
            total_deleted += 1
        if SLEEP_BETWEEN_BATCHES_SEC:
            time.sleep(SLEEP_BETWEEN_BATCHES_SEC)
    return total_deleted

def main():
    cutoff = datetime.now(timezone.utc) - timedelta(days=RETENTION_DAYS_FOR_RUNS)
    with SyncPrefectClient() as client:
        print(f"Start cleanup: runs<{RETENTION_DAYS_FOR_RUNS}d; cutoff={cutoff.isoformat()}; DRY_RUN={DRY_RUN}")
        total_runs_deleted = 0
        total_artifacts_deleted = 0

        while True:
            runs = client.read_flow_runs(
                flow_run_filter=FlowRunFilter(
                    start_time=FlowRunFilterStartTime(before_=cutoff),
                    state=FlowRunFilterState(
                        type=FlowRunFilterStateType.any_(END_STATES)
                    ),
                ),
                limit=BATCH_SIZE,
                offset=0,
            )
            if not runs:
                break

            for r in runs:
                total_artifacts_deleted += delete_artifacts_for_run(client, r.id)
                if DRY_RUN:
                    print(f"[DRY] would delete flow run {r.id} (start={getattr(r, 'start_time', None)})")
                else:
                    client.delete_flow_run(r.id)
                total_runs_deleted += 1

            if SLEEP_BETWEEN_BATCHES_SEC:
                time.sleep(SLEEP_BETWEEN_BATCHES_SEC)

        print(f"Done. runs_deleted={total_runs_deleted}, artifacts_deleted={total_artifacts_deleted}, DRY_RUN={DRY_RUN}")

if __name__ == "__main__":
    main()

-- events 时间列
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_event_occurred ON event (occurred);
-- logs 时间列
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_ts ON log ("timestamp");
-- flow/task 依赖键
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_task_run_flow ON task_run (flow_run_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_task_state_task ON task_run_state (task_run_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_flow_run_end ON flow_run (end_time);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_flow_state_flow ON flow_run_state (flow_run_id);

-- 重复执行，直到 0 行被删
WITH to_delete AS (
  SELECT ctid
  FROM event
  WHERE occurred < now() - interval '15 days'
  LIMIT 100000
)
DELETE FROM event e
USING to_delete d
WHERE e.ctid = d.ctid;

-- 删掉不再被任何 event 引用的资源行
DELETE FROM event_resources er
WHERE NOT EXISTS (
  SELECT 1 FROM event e WHERE e.resource_id = er.id
);

WITH to_delete AS (
  SELECT ctid
  FROM log
  WHERE "timestamp" < now() - interval '15 days'
  LIMIT 100000
)
DELETE FROM log l
USING to_delete d
WHERE l.ctid = d.ctid;

-- 先挑出旧的 flow_run（分批）
WITH old_flows AS (
  SELECT id
  FROM flow_run
  WHERE end_time IS NOT NULL
    AND end_time < now() - interval '15 days'
  LIMIT 50000
)
-- 如还没清 logs，可先清这些 runs 的日志
DELETE FROM log
WHERE flow_run_id IN (SELECT id FROM old_flows);

-- 先删 task_run_state
DELETE FROM task_run_state
WHERE task_run_id IN (
  SELECT id FROM task_run
  WHERE flow_run_id IN (SELECT id FROM old_flows)
);

-- 再删 task_run
DELETE FROM task_run
WHERE flow_run_id IN (SELECT id FROM old_flows);

-- 再删 flow_run_state
DELETE FROM flow_run_state
WHERE flow_run_id IN (SELECT id FROM old_flows);

-- 最后删 flow_run
DELETE FROM flow_run
WHERE id IN (SELECT id FROM old_flows);

VACUUM (ANALYZE) event;
VACUUM (ANALYZE) event_resources;
VACUUM (ANALYZE) log;
VACUUM (ANALYZE) task_run_state;
VACUUM (ANALYZE) task_run;
VACUUM (ANALYZE) flow_run_state;
VACUUM (ANALYZE) flow_run;

PREFECT_SERVER_EVENTS_RETENTION_PERIOD="P15D"

PREFECT_LOGGING_TO_API_ENABLED=false
# 或
PREFECT_LOGGING_LEVEL=WARNING

-- flow runs 及其关联（通常会级联 task runs / states）
VACUUM (ANALYZE) flow_run;
VACUUM (ANALYZE) flow_run_state;
VACUUM (ANALYZE) task_run;
VACUUM (ANALYZE) task_run_state;

-- 如果你也清了 logs / events
VACUUM (ANALYZE) log;
VACUUM (ANALYZE) event;
VACUUM (ANALYZE) event_resources;

-- 锁表强、影响写入；更推荐用 pg_repack 做在线重整
VACUUM FULL flow_run;
VACUUM FULL task_run;
VACUUM FULL task_run_state;
VACUUM FULL flow_run_state;

VACUUM (ANALYZE) flow_run;
VACUUM (ANALYZE) flow_run_state;
VACUUM (ANALYZE) task_run;
VACUUM (ANALYZE) task_run_state;
VACUUM (ANALYZE) log;
VACUUM (ANALYZE) event;
VACUUM (ANALYZE) event_resources;

ALTER TABLE event
  SET (autovacuum_vacuum_scale_factor = 0.01,
       autovacuum_vacuum_threshold = 10000,
       autovacuum_analyze_scale_factor = 0.02);

ALTER TABLE log
  SET (autovacuum_vacuum_scale_factor = 0.01,
       autovacuum_vacuum_threshold = 10000,
       autovacuum_analyze_scale_factor = 0.02);

autovacuum = on
autovacuum_max_workers = 6
autovacuum_naptime = '10s'
autovacuum_vacuum_cost_limit = 2000

ALTER SYSTEM SET log_autovacuum_min_duration = '0';
SELECT pg_reload_conf();

-- 关注死元组和最近的自动/手动 VACUUM 时间
SELECT relname, n_dead_tup, last_autovacuum, last_vacuum, autovacuum_count, vacuum_count
FROM pg_stat_user_tables
ORDER BY n_dead_tup DESC
LIMIT 20;

-- 查看 VACUUM 进度
SELECT * FROM pg_stat_progress_vacuum;

-- 排查阻碍 VACUUM 的长事务/会话
SELECT pid, state, xact_start, query
FROM pg_stat_activity
WHERE state <> 'idle'
ORDER BY xact_start;

PREFECT_LOGGING_LEVEL=WARNING

BEGIN;

-- 基准时间点（15天）
-- 可改：interval '15 days'
WITH cutoff AS (
  SELECT NOW() - interval '15 days' AS ts
),

-- 需要删除的 flow_run
old_flow_runs AS (
  SELECT fr.id
  FROM flow_run fr, cutoff c
  WHERE COALESCE(fr.end_time, fr.created) < c.ts
),

-- 需要删除的 task_run（隶属于 old_flow_runs）
old_task_runs AS (
  SELECT tr.id
  FROM task_run tr
  JOIN old_flow_runs fr ON fr.id = tr.flow_run_id
),

-- 1) 解除可能的外键引用，避免删除冲突
-- 如果有 task_run 指向将被删除的子流（child_flow_run_id），先置空
nulled_child AS (
  UPDATE task_run t
  SET child_flow_run_id = NULL
  WHERE t.child_flow_run_id IN (SELECT id FROM old_flow_runs)
  RETURNING 1
),
-- 如果有 task_run 指向将被删除的父 task（parent_task_run_id），先置空
nulled_parent AS (
  UPDATE task_run t
  SET parent_task_run_id = NULL
  WHERE t.parent_task_run_id IN (SELECT id FROM old_task_runs)
  RETURNING 1
),
-- flow/task 当前状态外键（state_id）先置空，避免状态表删除时触发约束
nulled_task_state_ptr AS (
  UPDATE task_run t
  SET state_id = NULL
  WHERE t.id IN (SELECT id FROM old_task_runs)
  RETURNING 1
),
nulled_flow_state_ptr AS (
  UPDATE flow_run f
  SET state_id = NULL
  WHERE f.id IN (SELECT id FROM old_flow_runs)
  RETURNING 1
)

-- 2) 先删从表数据（状态、日志、工件、输入等）
-- task_run_state
DELETE FROM task_run_state s
USING old_task_runs r
WHERE s.task_run_id = r.id;

-- flow_run_state
DELETE FROM flow_run_state s
USING old_flow_runs r
WHERE s.flow_run_id = r.id;

-- 日志（可能很多，注意执行时间）
DELETE FROM "log" l
WHERE l.flow_run_id IN (SELECT id FROM old_flow_runs)
   OR l.task_run_id IN (SELECT id FROM old_task_runs);

-- 工件
DELETE FROM artifact a
WHERE a.flow_run_id IN (SELECT id FROM old_flow_runs)
   OR a.task_run_id IN (SELECT id FROM old_task_runs);

-- flow_run 输入
DELETE FROM flow_run_input i
USING old_flow_runs r
WHERE i.flow_run_id = r.id;

-- 如有其他引用 run 的表（例如自定义扩展），请在此位置追加 DELETE

-- 3) 删除主表
DELETE FROM task_run t
USING old_task_runs r
WHERE t.id = r.id;

DELETE FROM flow_run f
USING old_flow_runs r
WHERE f.id = r.id;

COMMIT;

可选：用 pg_cron 定时

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_flow_run_end_time_notnull
  ON flow_run (end_time)
  WHERE end_time IS NOT NULL;

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_task_run_flow_run_id
  ON task_run (flow_run_id);

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_task_run_state_task_run_id
  ON task_run_state (task_run_id);

WITH cutoff AS (
  SELECT now() - interval '15 days' AS ts
),
to_delete_task_runs AS (
  SELECT tr.id
  FROM task_run tr
  JOIN flow_run fr ON fr.id = tr.flow_run_id
  JOIN cutoff c ON true
  WHERE fr.end_time IS NOT NULL
    AND fr.end_time < c.ts
  ORDER BY tr.id
  LIMIT 30000  -- 分批删除 3万条 task_run 对应的状态
)
DELETE FROM task_run_state s
USING to_delete_task_runs d
WHERE s.task_run_id = d.id;

-- 第一步：预计算本批要删的 task_run
CREATE TEMP TABLE tmp_old_task_runs ON COMMIT DROP AS
SELECT tr.id
FROM task_run tr
JOIN flow_run fr ON fr.id = tr.flow_run_id
WHERE fr.end_time IS NOT NULL
  AND fr.end_time < now() - interval '15 days'
ORDER BY tr.id
LIMIT 30000;

-- 第二步：基于临时表删除状态
DELETE FROM task_run_state s
USING tmp_old_task_runs t
WHERE s.task_run_id = t.id;

ANALYZE task_run_state;

BEGIN;

-- 1) 冻结本批要删除的 flow_run（15 天前完成）
CREATE TEMP TABLE _del_flow_runs ON COMMIT DROP AS
SELECT id
FROM flow_run
WHERE end_time IS NOT NULL
  AND end_time < now() - interval '15 days'
ORDER BY id
LIMIT 30000;  -- 分批量

-- 没有可删则退出
DO $$
BEGIN
  IF (SELECT COUNT(*) FROM _del_flow_runs) = 0 THEN
    RAISE NOTICE 'no flow_runs to delete';
  END IF;
END$$;

-- 2) 冻结本批相关的 task_run
CREATE TEMP TABLE _del_task_runs ON COMMIT DROP AS
SELECT tr.id
FROM task_run tr
JOIN _del_flow_runs d ON d.id = tr.flow_run_id;

-- 3) 先解除可能的外键指针，避免约束冲突（若你的 schema 无这些列，可略过）
UPDATE task_run t
SET child_flow_run_id = NULL
WHERE t.child_flow_run_id IN (SELECT id FROM _del_flow_runs);

UPDATE task_run t
SET parent_task_run_id = NULL
WHERE t.parent_task_run_id IN (SELECT id FROM _del_task_runs);

UPDATE task_run t
SET state_id = NULL
WHERE t.id IN (SELECT id FROM _del_task_runs);

UPDATE flow_run f
SET state_id = NULL
WHERE f.id IN (SELECT id FROM _del_flow_runs);

-- 4) 删从表（状态、日志、工件、输入等）
DELETE FROM task_run_state s
USING _del_task_runs t
WHERE s.task_run_id = t.id;

DELETE FROM flow_run_state s
USING _del_flow_runs f
WHERE s.flow_run_id = f.id;

DELETE FROM "log" l
WHERE l.flow_run_id IN (SELECT id FROM _del_flow_runs)
   OR l.task_run_id IN (SELECT id FROM _del_task_runs);

DELETE FROM artifact a
WHERE a.flow_run_id IN (SELECT id FROM _del_flow_runs)
   OR a.task_run_id IN (SELECT id FROM _del_task_runs);

DELETE FROM flow_run_input i
USING _del_flow_runs f
WHERE i.flow_run_id = f.id;

-- 5) 删主表
DELETE FROM task_run t
USING _del_task_runs dt
WHERE t.id = dt.id;

DELETE FROM flow_run f
USING _del_flow_runs df
WHERE f.id = df.id;

COMMIT;

SELECT
  tc.table_name, ccu.table_name AS referenced_table, rc.delete_rule
FROM information_schema.table_constraints AS tc
JOIN information_schema.referential_constraints rc
  ON rc.constraint_name = tc.constraint_name AND rc.constraint_schema = tc.table_schema
JOIN information_schema.constraint_column_usage AS ccu
  ON ccu.constraint_name = tc.constraint_name AND ccu.constraint_schema = tc.table_schema
WHERE tc.constraint_type = 'FOREIGN KEY'
  AND tc.table_schema = 'public'
  AND (tc.table_name, ccu.table_name) IN (
    ('task_run','flow_run'),
    ('task_run_state','task_run'),
    ('flow_run_state','flow_run')
  );

-- 查关键外键的 delete_rule（CASCADE / SET NULL / NO ACTION）
SELECT
  tc.table_name AS fk_table,
  kcu.column_name AS fk_column,
  ccu.table_name AS pk_table,
  rc.delete_rule
FROM information_schema.table_constraints AS tc
JOIN information_schema.key_column_usage AS kcu
  ON kcu.constraint_name = tc.constraint_name AND kcu.constraint_schema = tc.constraint_schema
JOIN information_schema.referential_constraints AS rc
  ON rc.constraint_name = tc.constraint_name AND rc.constraint_schema = tc.constraint_schema
JOIN information_schema.constraint_column_usage AS ccu
  ON ccu.constraint_name = tc.constraint_name AND ccu.constraint_schema = tc.constraint_schema
WHERE tc.constraint_type = 'FOREIGN KEY'
  AND tc.table_schema = 'public'
  AND (tc.table_name, kcu.column_name) IN (
    ('task_run','flow_run_id'),
    ('task_run','parent_task_run_id'),
    ('task_run','child_flow_run_id'),
    ('task_run','state_id'),
    ('flow_run','state_id'),
    ('task_run_state','task_run_id'),
    ('flow_run_state','flow_run_id')
  )
ORDER BY fk_table, fk_column;

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_task_run_child_flow_run_id ON task_run (child_flow_run_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_task_run_parent_task_run_id ON task_run (parent_task_run_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_task_run_state_id ON task_run (state_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_flow_run_state_id ON flow_run (state_id);

-- 准备索引（只需建一次；没有这个索引会很慢）
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_flow_run_end_time_notnull
  ON flow_run (end_time) WHERE end_time IS NOT NULL;

-- 按批次删除 3 万个超过 15 天的 flow_run，其它表走级联
WITH del AS (
  SELECT id
  FROM flow_run
  WHERE end_time IS NOT NULL
    AND end_time < now() - interval '15 days'
  ORDER BY id
  LIMIT 30000
)
DELETE FROM flow_run f
USING del d
WHERE f.id = d.id;

-- 可循环执行，直到受影响行数为 0

-- 按 end_time 过滤 flow_run（部分索引，避免扫描 NULL）
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_flow_run_end_time_notnull
  ON flow_run (end_time) WHERE end_time IS NOT NULL;

-- 连接 task_run → flow_run（有外键也建议有索引）
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_task_run_flow_run_id
  ON task_run (flow_run_id);

-- 日志按 run 关联删除（强烈建议）
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_flow_run_id
  ON "log" (flow_run_id);

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_task_run_id
  ON "log" (task_run_id);

-- 可选：工件、flow_run_input（若存在这些表，建议加索引）
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_artifact_flow_run_id
  ON artifact (flow_run_id);

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_artifact_task_run_id
  ON artifact (task_run_id);

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_flow_run_input_flow_run_id
  ON flow_run_input (flow_run_id);

BEGIN;

-- 1) 冻结本批要删除的 flow_run
CREATE TEMP TABLE _del_flow_runs ON COMMIT DROP AS
SELECT id
FROM flow_run
WHERE end_time IS NOT NULL
  AND end_time < now() - interval '15 days'
ORDER BY id
LIMIT 30000;  -- 调整批量大小以平衡时长与锁

-- 2) 冻结本批关联的 task_run（用于日志/工件等非级联表清理）
CREATE TEMP TABLE _del_task_runs ON COMMIT DROP AS
SELECT tr.id
FROM task_run tr
JOIN _del_flow_runs d ON d.id = tr.flow_run_id;

-- 3) 先清理日志（非级联）
DELETE FROM "log" l
WHERE l.flow_run_id IN (SELECT id FROM _del_flow_runs)
   OR l.task_run_id IN (SELECT id FROM _del_task_runs);

-- 4) 可选：清理工件（如存在该表）
DELETE FROM artifact a
WHERE a.flow_run_id IN (SELECT id FROM _del_flow_runs)
   OR a.task_run_id IN (SELECT id FROM _del_task_runs);

-- 5) 可选：清理 flow_run_input（如存在该表）
DELETE FROM flow_run_input i
USING _del_flow_runs d
WHERE i.flow_run_id = d.id;

-- 6) 删除 flow_run（将自动级联删除 task_run / flow_run_state / task_run_state）
DELETE FROM flow_run f
USING _del_flow_runs d
WHERE f.id = d.id;

COMMIT;

-- 如未启用
-- CREATE EXTENSION IF NOT EXISTS pg_cron;

-- 每天 03:00 开始，每 10 分钟跑一次，共 6 次（示意）
SELECT cron.schedule('prefect_cleanup_03_00', '0 3 * * *', $$BEGIN;
CREATE TEMP TABLE _del_flow_runs ON COMMIT DROP AS
SELECT id FROM flow_run WHERE end_time IS NOT NULL AND end_time < now() - interval '15 days'
ORDER BY id LIMIT 30000;
CREATE TEMP TABLE _del_task_runs ON COMMIT DROP AS
SELECT tr.id FROM task_run tr JOIN _del_flow_runs d ON d.id = tr.flow_run_id;
DELETE FROM "log" l WHERE l.flow_run_id IN (SELECT id FROM _del_flow_runs) OR l.task_run_id IN (SELECT id FROM _del_task_runs);
DELETE FROM artifact a WHERE a.flow_run_id IN (SELECT id FROM _del_flow_runs) OR a.task_run_id IN (SELECT id FROM _del_task_runs);
DELETE FROM flow_run_input i USING _del_flow_runs d WHERE i.flow_run_id = d.id;
DELETE FROM flow_run f USING _del_flow_runs d WHERE f.id = d.id;
COMMIT;$$);

SELECT cron.schedule('prefect_cleanup_03_10', '10 3 * * *', $$...同上脚本...$$);
-- 依次 03:20、03:30、03:40、03:50 再调度几次

-- 假设日志表时间列名为 created（请先确认你的列名）
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_created ON "log"(created);

DELETE FROM "log"
WHERE created < now() - interval '15 days'
LIMIT 500000;  -- 可选：分批删除，具体可用 CTE + ORDER BY + LIMIT

-- 两个都试一下，哪个不是 NULL 就说明实际注册的表名
SELECT to_regclass('public.log');
SELECT to_regclass('public."log"');

-- psql 里也可以
-- \d log
-- \d "log"

DELETE FROM log l
WHERE l.flow_run_id IN (SELECT id FROM _del_flow_runs)
   OR l.task_run_id IN (SELECT id FROM _del_task_runs);

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_flow_run_id_notnull
  ON log (flow_run_id) WHERE flow_run_id IS NOT NULL;

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_task_run_id_notnull
  ON log (task_run_id) WHERE task_run_id IS NOT NULL;

DELETE FROM log l
WHERE l.flow_run_id IN (SELECT id FROM _del_flow_runs)
   OR l.task_run_id IN (SELECT id FROM _del_task_runs);

-- 先按 flow_run_id 删
DELETE FROM log l
USING _del_flow_runs d
WHERE l.flow_run_id = d.id;

-- 再按 task_run_id 删
DELETE FROM log l
USING _del_task_runs t
WHERE l.task_run_id = t.id;

DO $$
DECLARE
  batch_limit integer := 100000;
  rows_deleted integer;
BEGIN
  -- 先针对 flow_run_id 的日志
  LOOP
    WITH to_del AS (
      SELECT l.ctid
      FROM log l
      JOIN _del_flow_runs d ON l.flow_run_id = d.id
      LIMIT batch_limit
    )
    DELETE FROM log WHERE ctid IN (SELECT ctid FROM to_del);

    GET DIAGNOSTICS rows_deleted = ROW_COUNT;
    EXIT WHEN rows_deleted = 0;
  END LOOP;

  -- 再针对 task_run_id 的日志
  LOOP
    WITH to_del AS (
      SELECT l.ctid
      FROM log l
      JOIN _del_task_runs t ON l.task_run_id = t.id
      LIMIT batch_limit
    )
    DELETE FROM log WHERE ctid IN (SELECT ctid FROM to_del);

    GET DIAGNOSTICS rows_deleted = ROW_COUNT;
    EXIT WHEN rows_deleted = 0;
  END LOOP;
END$$;

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_created ON log (created);

-- 微批按时间删（Postgres 不允许 DELETE … ORDER BY）
DO $$
DECLARE
  rows_deleted integer;
BEGIN
  LOOP
    WITH to_del AS (
      SELECT ctid
      FROM log
      WHERE created < now() - interval '15 days'
      LIMIT 100000
    )
    DELETE FROM log WHERE ctid IN (SELECT ctid FROM to_del);
    GET DIAGNOSTICS rows_deleted = ROW_COUNT;
    EXIT WHEN rows_deleted = 0;
  END LOOP;
END$$;

ANALYZE log;

my_task()

from prefect import flow, task, get_run_logger

@task
def may_fail(x):
    if x == 2:
        raise RuntimeError(f"blade_req_final_failed=x={x}")
    return x * 10

@flow
def my_flow():
    logger = get_run_logger()
    try:
        a = may_fail(1)  # 正常
        b = may_fail(2)  # 这里抛出异常，直接被下面的 except 捕获
        c = may_fail(3)  # 不会执行到
    except Exception as exc:
        logger.error(f"Flow 捕获到 task 异常: {exc}", exc_info=True)
        # 需要让 Flow 标记为失败就 re-raise；只记录不失败就不 raise
        raise

if __name__ == "__main__":
    my_flow()

task.submit(...)

future.result()

wait/as_completed

from prefect import flow, task, get_run_logger
from prefect.futures import as_completed

@task
def may_fail(x):
    if x == 2:
        raise RuntimeError(f"blade_req_final_failed=x={x}")
    return x * 10

@flow
def my_flow():
    logger = get_run_logger()
    futs = [may_fail.submit(i) for i in range(5)]
    try:
        # 任意一个 future 在 .result() 时失败，都会被这里的 except 捕获
        for fut in as_completed(futs):
            _ = fut.result()  # 默认 raise_on_failure=True
    except Exception as exc:
        logger.error(f"Flow 捕获到 task 异常: {exc}", exc_info=True)
        # 想让 Flow 失败就继续 raise
        raise

from prefect import flow, task, get_run_logger
from prefect.futures import wait
from prefect.states import get_state_exception

@task
def may_fail(x):
    if x % 3 == 0:
        raise RuntimeError(f"blade_req_final_failed=x={x}")
    return x * 10

@flow
def my_flow():
    logger = get_run_logger()
    futs = [may_fail.submit(i) for i in range(1, 10)]

    # 等待所有 future 完成
    done, _ = wait(futs)

    failed = []
    for fut in done:
        st = fut.state
        if st.is_failed():
            exc = get_state_exception(st)  # 拿到原始异常
            logger.error(f"Task 失败: task_run_id={fut.task_run_id}, exc={exc}", exc_info=True)
            failed.append(exc)

    if failed:
        # 如需让 Flow 标记为失败，可以合并/抛出一个代表性异常
        raise RuntimeError(f"{len(failed)} task(s) failed; see logs above.")

future.state

get_state_exception(state)

get_run_logger()

on_failure

.submit()

submit()

wait()

future.state

from prefect import flow, get_run_logger
from prefect.context import get_run_context
from prefect.futures import wait
from prefect.states import get_state_exception
import traceback

@flow(flow_run_name="{logid}")
def my_flow(input_data, logid):
    logger = get_run_logger()
    mario_log.notice(f"act=my_flow_run, logid={logid}")

    flow_run_id = get_run_context().flow_run.id

    try:
        # 并发提交
        a = task_a.submit(input_data)
        b = task_b.with_options(retries=3).submit(a)
        c = task_c.submit(a)

        tasks = {"b": b, "c": c}

        # 用 a 的结果做条件判断；如果 a 失败，这里会抛出并被外层 except 捕获
        a_value = a.result()
        if a_value["value"] > 30:
            e = task_e.submit(a)
            tasks["e"] = e

        d = task_d.submit(tasks)

        # 等待所有已提交任务完成，并逐个检查失败，打印具体是哪个 task 挂了
        name_to_future = {"a": a, **tasks, "d": d}
        wait(list(name_to_future.values()))

        any_failed = False
        for name, fut in name_to_future.items():
            st = fut.state
            if st.is_failed():
                exc = get_state_exception(st)
                mario_log.warning(f"act=task_failed, logid={logid}, task={name}, exc={exc}")
                logger.error(f"task={name} failed: {exc}", exc_info=True)
                any_failed = True

        if any_failed:
            # 统一走到 except 做失败返回
            raise RuntimeError("One or more tasks failed, see logs above.")

        # 所有任务成功，返回最终结果
        return {"flow_run_id": flow_run_id, "errno": 0, "errmsg": "", "result": d.result()}

    except Exception as e:
        mario_log.warning(f"act=run_flow_failed, logid={logid}, traceback={traceback.format_exc()}")
        logger.error(f"Flow failed: {e}", exc_info=True)
        return {"flow_run_id": flow_run_id, "errno": 1, "errmsg": str(e)}

wait(...)

future.state.is_failed()

get_state_exception(state)

a.state

d

d.result()

a

task_e

from prefect import flow
from prefect.context import get_run_context
import traceback

@flow(flow_run_name="{logid}")
def my_flow(input_data, logid):
    """ my_flow """
    mario_log.notice(f"act=my_flow_run, logid={logid}")

    flow_run_id = get_run_context().flow_run.id

    try:
        a = task_a.submit(input_data)

        b = task_b.with_options(retries=3).submit(a)
        c = task_c.submit(a)

        # 关键：总是提交 e，但把“是否需要执行”的判断放进 task_e 里
        # 如果不满足条件，task_e 自己直接 no-op/返回 None
        e = task_e.submit(a)

        tasks = {'b': b, 'c': c, 'e': e}

        d = task_d.submit(tasks)

        # 只在最后取一次结果；任何上游失败都会在这里抛出异常
        result = d.result()

        return {"flow_run_id": flow_run_id, "errno": 0, "errmsg": "", "result": result}
    except Exception as e:
        mario_log.warning(f"act=run_flow_failed, logid={logid}, traceback={traceback.format_exc()}")
        return {"flow_run_id": flow_run_id, "errno": 1, "errmsg": str(e)}

task_e

from prefect import task

@task
def task_e(a_value):
    # a_value 会由 Prefect 自动解析为 task_a 的结果
    if a_value["value"] <= 30:
        return None  # 不需要执行时直接返回
    # 否则执行你的原逻辑
    # ... 原有 task_e 的实现 ...

d

.result()

from prefect.futures import resolve_futures_to_results
resolve_futures_to_results([d])  # 任一上游失败会在这里抛出
result = d.result()

d.result()

resolve_futures_to_states(...)

get_state_exception(state)

from prefect import flow, get_run_logger
from prefect.context import get_run_context
from prefect.futures import resolve_futures_to_states
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback
import traceback

# 可选：task_e 内部做条件判断，不满足则直接 no-op
# @task
# def task_e(a_value):
#     if a_value["value"] <= 30:
#         return None
#     ...  # 你的逻辑

@flow(flow_run_name="{logid}")
def my_flow(input_data, logid):
    """ my_flow """
    logger = get_run_logger()
    mario_log.notice(f"act=my_flow_run, logid={logid}")

    flow_run_id = get_run_context().flow_run.id

    try:
        a = task_a.submit(input_data)
        b = task_b.with_options(retries=3).submit(a)
        c = task_c.submit(a)
        e = task_e.submit(a)  # 是否需要执行在 task_e 内部判断

        tasks = {"b": b, "c": c, "e": e}
        d = task_d.submit(tasks)

        # 只在这里取一次结果；任何上游失败都会在这里抛出
        result = d.result()

        return {"flow_run_id": flow_run_id, "errno": 0, "errmsg": "", "result": result}

    except Exception as err:
        # 统一失败日志
        mario_log.warning(f"act=run_flow_failed, logid={logid}, traceback={traceback.format_exc()}")
        logger.error(f"Flow failed: {err}", exc_info=True)

        # 一次性解析所有相关 futures -> states，然后把失败根因全部打印出来
        try:
            name_to_future = {"a": a, **tasks, "d": d}
            states = resolve_futures_to_states(name_to_future)
            for name, st in states.items():
                if st.is_failed():
                    exc = get_state_exception(st)
                    mario_log.warning(
                        f"act=task_failed, logid={logid}, task={name}, exc={exc}, tb={exception_traceback(exc)}"
                    )
                    logger.error(f"task={name} failed: {exc}\n{exception_traceback(exc)}")
        except Exception:
            # 即使解析/打印失败也不影响主返回
            pass

        return {"flow_run_id": flow_run_id, "errno": 1, "errmsg": str(err)}

resolve_futures_to_states(...)

get_state_exception(state)

raise

on_failure

get_state_exception(state)

with_options

from prefect import task, flow, get_run_logger
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback

def log_task_failure(task, task_run, state):
    # 这里不会重试，说明已是最终失败
    exc = get_state_exception(state)
    logger = get_run_logger()
    # 你也可以用 mario_log
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )

@task(on_failure=[log_task_failure])
def task_a(input_data):
    ...

# 如果你不想修改原装饰器，也可以在使用处挂钩子
# task_b = task_b.with_options(on_failure=[log_task_failure])

from prefect.states import get_state_exception

def log_flow_failure(flow, flow_run, state):
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[FLOW FAILED] flow={flow.name} flow_run_id={flow_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )

@flow(flow_run_name="{logid}", on_failure=[log_flow_failure])
def my_flow(input_data, logid):
    ...

task_e

d.result()

from prefect import flow, get_run_logger
from prefect.context import get_run_context
import traceback

def log_task_failure(task, task_run, state):
    from prefect.states import get_state_exception
    from prefect.exceptions import exception_traceback
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )
    # 如需同步打到你的日志系统：
    # mario_log.warning(f"act=task_failed, logid={task_run.id}, task={task.name}, exc={exc}")

@task(on_failure=[log_task_failure])
def task_a(input_data):
    ...

@task(on_failure=[log_task_failure])
def task_b(a_value):
    ...

@task(on_failure=[log_task_failure])
def task_c(a_value):
    ...

@task(on_failure=[log_task_failure])
def task_e(a_value):
    if a_value["value"] <= 30:
        return None
    # 否则执行你的逻辑
    ...

@task(on_failure=[log_task_failure])
def task_d(tasks_dict):
    ...

@flow(flow_run_name="{logid}")
def my_flow(input_data, logid):
    logger = get_run_logger()
    mario_log.notice(f"act=my_flow_run, logid={logid}")

    flow_run_id = get_run_context().flow_run.id

    try:
        a = task_a.submit(input_data)
        b = task_b.with_options(retries=3).submit(a)
        c = task_c.submit(a)
        e = task_e.submit(a)

        tasks = {'b': b, 'c': c, 'e': e}
        d = task_d.submit(tasks)

        result = d.result()  # 任一上游失败会在此抛出，上面的 on_failure 已经记录过细节
        return {"flow_run_id": flow_run_id, "errno": 0, "errmsg": "", "result": result}

    except Exception as e:
        # Flow 级兜底日志（简单化即可，详细根因已由 task 的 on_failure 打过）
        logger.error(f"Flow failed: {e}", exc_info=True)
        mario_log.warning(f"act=run_flow_failed, logid={logid}, traceback={traceback.format_exc()}")
        return {"flow_run_id": flow_run_id, "errno": 1, "errmsg": str(e)}

task_run.parameters

get_run_context()

prefect.task

task

from functools import partial
from prefect import task as _task, get_run_logger
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback

def log_task_failure(task, task_run, state):
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )

# 重绑定：给 task 预设默认的 on_failure 钩子
task = partial(_task, on_failure=[log_task_failure])

# 之后在本模块里，直接用 @task，就会自动带上 on_failure
@task
def task_a(x):
    ...

# 需要额外 options 也可以叠加传递
@task(retries=3, retry_delay_seconds=5)
def task_b(x):
    ...

from my_prefect import task

from prefect import task as _task, get_run_logger
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback

def log_task_failure(task, task_run, state):
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )

def monitored_task(__fn=None, on_failure=None, **opts):
    # 合并默认钩子 + 额外钩子
    hooks = [log_task_failure] + (on_failure or [])

    def _decorate(fn):
        return _task(on_failure=hooks, **opts)(fn)

    # 既支持 @monitored_task，也支持 @monitored_task(retries=3) 用法
    return _decorate if __fn is None else _decorate(__fn)

# 用法示例
@monitored_task
def task_a(x):
    ...

@monitored_task(retries=3)
def task_b(x):
    ...

def extra_hook(task, task_run, state): ...
@monitored_task(on_failure=[extra_hook])
def task_c(x):
    ...

.with_options(on_failure=[log_task_failure])

from other.module import raw_task_b as _raw_task_b
    task_b = _raw_task_b.with_options(on_failure=[log_task_failure])

from functools import partial
  from prefect import flow as _flow
  flow = partial(_flow, on_failure=[log_flow_failure])

on_failure

from prefect import task as _prefect_task, get_run_logger
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback
from prefect.cache_policies import NO_CACHE

def log_task_failure(task, task_run, state):
    # 统一的失败钩子：打印原始异常 + traceback
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )

def task(__fn=None, **kwargs):
    # 统一默认项
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)

    # 合并 on_failure：默认钩子 + 用户传入（可为单个或列表）
    default_hooks = [log_task_failure]
    user_hooks = kwargs.pop("on_failure", None)
    if user_hooks is None:
        hooks = default_hooks
    elif isinstance(user_hooks, (list, tuple)):
        hooks = default_hooks + list(user_hooks)
    else:
        hooks = default_hooks + [user_hooks]
    kwargs["on_failure"] = hooks

    def _decorate(fn):
        return _prefect_task(**kwargs)(fn)

    # 支持 @task 和 @task(...)
    return _decorate if __fn is None else _decorate(__fn)

@task
def task_a(x):
    ...

@task(retries=3)
def task_b(x):
    ...

def extra_hook(task, task_run, state): ...
@task(on_failure=[extra_hook], timeout_seconds=30)
def task_c(x):
    ...

kwargs.setdefault("on_failure", log_task_failure)

NO_CACHE

prefect.cache_policies

persist_result

@task

self.method.submit(...)

ClassName.method.submit(self, ...)

class Svc:
    @task  # 你的带默认 on_failure 的包装
    def step(self, x):
        ...

svc = Svc()
# 不要用 svc.step.submit(x)
# 要这样：
Svc.step.submit(svc, x)  # 显式把 self 作为第一个参数传进去

self

from prefect import task as _prefect_task, get_run_logger
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback
from prefect.cache_policies import NO_CACHE

def log_task_failure(task, task_run, state):
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )

class _BoundTaskProxy:
    def __init__(self, task_obj, instance):
        self._task = task_obj
        self._instance = instance
    # 直接调用：self.step(...)
    def __call__(self, *args, **kwargs):
        return self._task(self._instance, *args, **kwargs)
    # 并发提交：self.step.submit(...)
    def submit(self, *args, **kwargs):
        return self._task.submit(self._instance, *args, **kwargs)
    # 其他属性/方法透传
    def __getattr__(self, name):
        return getattr(self._task, name)

class _MethodFriendlyDescriptor:
    def __init__(self, task_obj):
        self._task = task_obj
    def __get__(self, obj, owner):
        # 通过类访问：返回原始 Task（适合 Class.method.submit(self, ...) 用法）
        if obj is None:
            return self._task
        # 通过实例访问：返回自动注入 self 的代理（支持 self.method.submit(...)）
        return _BoundTaskProxy(self._task, obj)

def task(__fn=None, **kwargs):
    # 统一默认项
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)

    # 合并 on_failure：默认 + 用户
    default_hooks = [log_task_failure]
    user_hooks = kwargs.pop("on_failure", None)
    if user_hooks is None:
        hooks = default_hooks
    elif isinstance(user_hooks, (list, tuple)):
        hooks = default_hooks + list(user_hooks)
    else:
        hooks = default_hooks + [user_hooks]
    kwargs["on_failure"] = hooks

    def _decorate(fn):
        t = _prefect_task(**kwargs)(fn)
        return _MethodFriendlyDescriptor(t)

    return _decorate if __fn is None else _decorate(__fn)

class Svc:
    @task
    def step(self, x):
        return x * 2

svc = Svc()

# 现在两种写法都可以：
r1 = svc.step.submit(10)         # 自动注入 self
r2 = Svc.step.submit(svc, 10)    # 显式传 self

@staticmethod

class Svc:
    @staticmethod
    @task
    def step(x, ctx):
        ...

@task

on_failure

setdefault("on_failure", [log_task_failure])

NO_CACHE

prefect.cache_policies

from prefect import task as _prefect_task, get_run_logger
from prefect.cache_policies import NO_CACHE
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback

# 统一失败钩子
def log_task_failure(task, task_run, state):
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )
    # 如果要同步写你们的日志系统：
    # import logger.log as mario_log
    # mario_log.warning(f"act=task_run_failed task={task.name} exc={exc}\n{exception_traceback(exc)}")


class _BoundTaskProxy:
    def __init__(self, task_obj, instance):
        self._task = task_obj
        self._instance = instance

    # 直接调用：self.step(...)
    def __call__(self, *args, **kwargs):
        return self._task(self._instance, *args, **kwargs)

    # 并发提交：self.step.submit(...)
    def submit(self, *args, **kwargs):
        return self._task.submit(self._instance, *args, **kwargs)

    # 支持 with_options：返回新的代理，继续保持自动注入 self
    def with_options(self, **opts):
        new_task = self._task.with_options(**opts)
        return _BoundTaskProxy(new_task, self._instance)

    # 其他属性/方法透传
    def __getattr__(self, name):
        return getattr(self._task, name)


class _MethodFriendlyDescriptor:
    def __init__(self, task_obj):
        self._task = task_obj

    def __get__(self, obj, owner):
        # 通过类访问：返回原始 Task（适合 Class.method.submit(self, ...) 用法）
        if obj is None:
            return self._task
        # 通过实例访问：返回自动注入 self 的代理（支持 self.method.submit(...)）
        return _BoundTaskProxy(self._task, obj)


def task(__fn=None, **kwargs):
    # 统一默认项
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)

    # 合并 on_failure：默认钩子 + 用户传入（可单个或列表）
    default_hooks = [log_task_failure]
    user_hooks = kwargs.pop("on_failure", None)
    if user_hooks is None:
        hooks = default_hooks
    elif isinstance(user_hooks, (list, tuple)):
        hooks = default_hooks + list(user_hooks)
    else:
        hooks = default_hooks + [user_hooks]
    kwargs["on_failure"] = hooks

    def _decorate(fn):
        t = _prefect_task(**kwargs)(fn)
        return _MethodFriendlyDescriptor(t)

    # 支持 @task 和 @task(...)
    return _decorate if __fn is None else _decorate(__fn)

class Svc:
    @task  # 自动带默认 on_failure
    def step(self, x):
        if x == 2:
            raise RuntimeError("blade_req_final_failed=x=2")
        return x * 2

svc = Svc()

# 方式 1：实例上直接 submit（自动注入 self）
f1 = svc.step.submit(2)

# 方式 2：类上 submit（显式传 self）
f2 = Svc.step.submit(svc, 2)

# 需要额外 options 时
f3 = svc.step.with_options(retries=3).submit(3)

kwargs

_state_

from prefect import task as _prefect_task, get_run_logger
from prefect.cache_policies import NO_CACHE
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback

# 默认失败钩子
def log_task_failure(task, task_run, state):
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )
    # 如需你们的日志：import logger.log as mario_log; mario_log.warning(...)

class _BoundTaskProxy:
    def __init__(self, task_obj, instance):
        self._task = task_obj
        self._instance = instance

    def __call__(self, *args, **kwargs):
        return self._task(self._instance, *args, **kwargs)

    def submit(self, *args, **kwargs):
        return self._task.submit(self._instance, *args, **kwargs)

    def with_options(self, **opts):
        new_task = self._task.with_options(**opts)
        return _BoundTaskProxy(new_task, self._instance)

    def __getattr__(self, name):
        return getattr(self._task, name)

class _MethodFriendlyDescriptor:
    def __init__(self, task_obj):
        self._task = task_obj
    def __get__(self, obj, owner):
        if obj is None:
            return self._task        # 类访问：BladesTask.req
        return _BoundTaskProxy(self._task, obj)  # 实例访问：blades_client.req

def task(__fn=None, **kwargs):
    # 统一默认项
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)

    # 合并默认 + 用户 on_failure
    default_hooks = [log_task_failure]
    user_hooks = kwargs.pop("on_failure", None)
    if user_hooks is None:
        hooks = default_hooks
    elif isinstance(user_hooks, (list, tuple)):
        hooks = default_hooks + list(user_hooks)
    else:
        hooks = default_hooks + [user_hooks]
    kwargs["on_failure"] = hooks

    def _decorate(fn):
        t = _prefect_task(**kwargs)(fn)
        return _MethodFriendlyDescriptor(t)

    return _decorate if __fn is None else _decorate(__fn)

from my_prefect_wrappers import task

class BladesTask:
    @task  # 自动带 on_failure=[log_task_failure]
    def req(self, x):
        if x == "boom":
            raise RuntimeError(f"blade_req_final_failed=x={x}")
        return {"ok": x}

blades_client = BladesTask()

# 你的用法：先取成员，再 submit
blades_req = blades_client.req   # 这是 _BoundTaskProxy
fut1 = blades_req.submit(123)
print(fut1.result())

# 或直接一行
fut2 = blades_client.req.submit("boom")  # 失败时会触发 log_task_failure
fut2.result()  # 这里会抛 UpstreamTaskError；根因已在钩子里打印过

__init__

blades_client.req

task(...)

@task

print(type(BladesTask.req), type(blades_client.req))
  # 应该显示 _MethodFriendlyDescriptor / _BoundTaskProxy（或内部包装类型）

req

from prefect import task as _prefect_task, get_run_logger
from prefect.cache_policies import NO_CACHE
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback

# 默认失败钩子
def log_task_failure(task, task_run, state):
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )
    # 如需你们的日志系统：
    # import logger.log as mario_log
    # mario_log.warning(f"act=task_run_failed task={task.name} exc={exc}\n{exception_traceback(exc)}")

class _BoundTaskProxy:
    def __init__(self, task_obj, instance):
        self._task = task_obj
        self._instance = instance

    def __call__(self, *args, **kwargs):
        return self._task(self._instance, *args, **kwargs)

    def submit(self, *args, **kwargs):
        return self._task.submit(self._instance, *args, **kwargs)

    def with_options(self, **opts):
        # 合并默认 on_failure + 用户 on_failure，避免覆盖默认钩子
        user_hooks = opts.pop("on_failure", None)
        if user_hooks is not None:
            base_default = getattr(self._task, "_default_on_failure_hooks", [])
            if isinstance(user_hooks, (list, tuple)):
                merged = list(base_default) + list(user_hooks)
            else:
                merged = list(base_default) + [user_hooks]
            opts["on_failure"] = merged

        new_task = self._task.with_options(**opts)
        return _BoundTaskProxy(new_task, self._instance)

    def __getattr__(self, name):
        return getattr(self._task, name)

class _MethodFriendlyDescriptor:
    def __init__(self, task_obj):
        self._task = task_obj
    def __get__(self, obj, owner):
        if obj is None:
            return self._task
        return _BoundTaskProxy(self._task, obj)

def task(__fn=None, **kwargs):
    # 默认项
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)

    # 合并默认 + 用户 on_failure
    default_hooks = [log_task_failure]
    user_hooks = kwargs.pop("on_failure", None)
    if user_hooks is None:
        hooks = default_hooks
    elif isinstance(user_hooks, (list, tuple)):
        hooks = default_hooks + list(user_hooks)
    else:
        hooks = default_hooks + [user_hooks]
    kwargs["on_failure"] = hooks

    def _decorate(fn):
        t = _prefect_task(**kwargs)(fn)
        # 标记默认钩子，供实例 with_options 合并使用
        t._default_on_failure_hooks = default_hooks
        return _MethodFriendlyDescriptor(t)

    return _decorate if __fn is None else _decorate(__fn)

raise RuntimeError("boom")

fut = blades_client.req.submit("x"); fut.result()

blades_client = BladesTask()
blades_req = blades_client.req   # 这是代理对象
blades_req.submit(xxx)

def bind_task(task_obj, instance):
    class _BoundTask:
        def __init__(self, t, inst):
            self._task = t
            self._inst = inst
        def submit(self, *args, **kwargs):
            return self._task.submit(self._inst, *args, **kwargs)
        def __call__(self, *args, **kwargs):
            return self._task(self._inst, *args, **kwargs)
        def with_options(self, **opts):
            new_t = self._task.with_options(**opts)
            return bind_task(new_t, self._inst)
        def __getattr__(self, name):
            return getattr(self._task, name)
    return _BoundTask(task_obj, instance)

class Svc:
    @task  # 这里可以是你带默认 on_failure 的包装，或原生 prefect.task
    def step(self, x):
        ...

svc = Svc()
blades_req = bind_task(Svc.step, svc)  # 预先绑定 self
blades_req.submit(xxx)

class Svc:
    @task
    def step(self, x):
        ...
    @property
    def blades_req(self):
        return bind_task(Svc.step, self)

svc = Svc()
svc.blades_req.submit(xxx)

state

task_run_name

__init__(..., _bns_=...)

from prefect import task as _prefect_task, get_run_logger
from prefect.cache_policies import NO_CACHE
from prefect.states import get_state_exception
from prefect.exceptions import exception_traceback

# 统一失败钩子：确保参数名和内部引用一致
def log_task_failure(task, task_run, state):
    exc = get_state_exception(state)
    logger = get_run_logger()
    logger.error(
        f"[TASK FAILED] task={task.name} run_id={task_run.id} "
        f"exc={exc}\n{exception_traceback(exc)}"
    )

class _BoundTaskProxy:
    def __init__(self, task_obj, instance):
        self._task = task_obj
        self._instance = instance

    def __call__(self, *args, **kwargs):
        return self._task(self._instance, *args, **kwargs)

    def submit(self, *args, **kwargs):
        return self._task.submit(self._instance, *args, **kwargs)

    def with_options(self, **opts):
        # 合并默认 on_failure（避免被覆盖）
        user_hooks = opts.pop("on_failure", None)
        if user_hooks is not None:
            base_default = getattr(self._task, "_default_on_failure_hooks", [])
            if isinstance(user_hooks, (list, tuple)):
                merged = list(base_default) + list(user_hooks)
            else:
                merged = list(base_default) + [user_hooks]
            opts["on_failure"] = merged

        new_task = self._task.with_options(**opts)
        return _BoundTaskProxy(new_task, self._instance)

    def __getattr__(self, name):
        return getattr(self._task, name)

class _MethodFriendlyDescriptor:
    def __init__(self, task_obj):
        self._task = task_obj
    def __get__(self, obj, owner):
        if obj is None:
            return self._task
        return _BoundTaskProxy(self._task, obj)

def task(__fn=None, **kwargs):
    # 默认项
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)

    # 合并默认 + 用户 on_failure
    default_hooks = [log_task_failure]
    user_hooks = kwargs.pop("on_failure", None)
    if user_hooks is None:
        hooks = default_hooks
    elif isinstance(user_hooks, (list, tuple)):
        hooks = default_hooks + list(user_hooks)
    else:
        hooks = default_hooks + [user_hooks]
    kwargs["on_failure"] = hooks

    def _decorate(fn):
        t = _prefect_task(**kwargs)(fn)
        t._default_on_failure_hooks = default_hooks
        return _MethodFriendlyDescriptor(t)

    return _decorate if __fn is None else _decorate(__fn)

使用方式（你想要的 blades_req.submit）：

from prefect import task

self.req

@task

如何在 Flow 里调用

说明
- bns client 的复用以前靠类成员，现在通过 `get_blades_bns_client` 的模块级单例实现。
- 你原代码里像 `_name_`、`_timeout_`、`_headers_`、`_content_` 这种参数名是无效的，已改回标准参数名，否则 Prefect/httpx 都不会识别。
- 外部依赖（如 `fs.save`、`mario_log`、`get_local_ip`）如果不是全局可用，建议作为参数注入或在函数内做 try/except 保护。

如果你不想用全局单例，也可以把 `bns_client` 作为参数传入

_decorate(_fn_)

fn

_fn_

__fn

log_task_failure

task/state

@task(...)

from typing import Optional, Callable
from prefect import task as prefect_task
from prefect.cache_policies import NO_CACHE
from prefect.states import get_state_exception, State
from prefect.exceptions import exception_traceback

def log_task_failure(task, task_run, state: State):
    """
    失败钩子：只会在最终失败时触发
    """
    exc = get_state_exception(state)
    tb = ""
    try:
        if exc:
            tb = exception_traceback(exc)
    except Exception:
        pass

    # 你自己的日志
    try:
        mario_log.warning(f"act=task_run_failed task={task.name} exc={exc}\n{tb}")
    except NameError:
        # 没有 mario_log 也不阻塞
        pass


def task(__fn: Optional[Callable] = None, **kwargs):
    """
    你的统一包装：支持 @task 和 @task(...)
    同时支持 sync/async 函数
    """
    # 统一默认值
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)
    kwargs.setdefault("on_failure", [log_task_failure])

    def _decorate(fn: Callable):
        # 直接委托给 Prefect 的官方装饰器（它本身支持 async）
        return prefect_task(**kwargs)(fn)

    # 支持两种用法：@task / @task(...)
    return _decorate(__fn) if callable(__fn) else _decorate

@task
async def my_async_task(x: int) -> int:
    await asyncio.sleep(0.1)
    return x + 1

@task(name="plus-two", retries=2)
def my_sync_task(x: int) -> int:
    return x + 2

await

.submit()

from prefect import flow

@flow
async def af():
    y = await my_async_task(1)
    z = my_sync_task(2)  # 同步 task 直接返回结果
    return y + z

@flow
def sf():
    z = my_sync_task(2)
    y = my_async_task(1)  # 在同步 flow 中，async task 会被引擎调度执行，直接返回结果
    return y + z

task

prefect.task(**kwargs)(fn)

on_failure

kwargs.setdefault

task

await

task

get_state_exception

prefect.states

prefect.exceptions

from typing import Callable, Optional

from prefect import task as prefect_task
from prefect.cache_policies import NO_CACHE
from prefect.states import State, get_state_exception
from prefect.exceptions import exception_traceback

def log_task_failure(task, task_run, state: State):
    """在最终失败时触发的 hook"""
    exc = get_state_exception(state)
    try:
        tb = exception_traceback(exc) if exc else ""
    except Exception:
        tb = ""
    try:
        mario_log.warning(f"act=task_run_failed task={task.name} exc={exc}\n{tb}")
    except NameError:
        # 没有 mario_log 也不阻塞
        pass

def task(__fn: Optional[Callable] = None, **kwargs):
    """自定义 task 装饰器：统一默认值 + 兼容 sync/async"""
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)
    kwargs.setdefault("on_failure", [log_task_failure])

    def _decorate(fn: Callable):
        return prefect_task(**kwargs)(fn)

    # 支持 @task 和 @task(...)
    return _decorate(__fn) if callable(__fn) else _decorate

await

from prefect import flow
import asyncio

@task
async def my_async_task(x: int) -> int:
    await asyncio.sleep(0.1)
    return x + 1

@flow
async def aflow():
    y = await my_async_task(1)  # 注意 await
    return y

from prefect import flow

@flow
def sflow():
    y = my_async_task(1)  # 不需要 await
    return y

# 在交互式/脚本里这么用
import asyncio
asyncio.run(my_async_task(1))

on_failure

from prefect.states import format_exception

def log_task_failure(task, task_run, state):
    mario_log.warning(
        f"act=task_run_failed task={task.name} {format_exception(state)}"
    )

import inspect
from prefect.states import get_state_exception, format_exception
from prefect.exceptions import exception_traceback

def log_task_failure(task, task_run, state):
    exc = None
    try:
        exc = get_state_exception(state)  # 有些环境下可能返回 coroutine
    except Exception:
        pass

    if isinstance(exc, BaseException) and not inspect.iscoroutine(exc):
        try:
            tb = exception_traceback(exc)
        except Exception:
            tb = ""
        msg = f"exc={exc}\n{tb}"
    else:
        # 没拿到 Exception（或拿到的是 coroutine），回退到字符串
        msg = format_exception(state)

    mario_log.warning(f"act=task_run_failed task={task.name} {msg}")

BaseException

on_failure

await

from prefect.states import get_state_exception, format_exception

from prefect.exceptions import exception_traceback

get_state_exception

task_run

task_run.name

task.name

from prefect.states import format_exception

def log_task_failure(task, task_run, state):
    run_name = getattr(task_run, "name", None) or "(unknown)"
    task_name = getattr(task, "name", None) or "(unknown)"
    msg = f"act=task_run_failed task={task_name} run_name={run_name} {format_exception(state)}"
    try:
        mario_log.warning(msg)
    except NameError:
        print(msg)

from prefect import task

@task(task_run_name="blades-{feat_id}-{cos_name}")
async def blades_req(feat_id, cos_name="req"):
    ...

from prefect.states import format_exception

def log_task_failure(task, task_run, state):
    run_name = getattr(task_run, "name", "(unknown)")
    msg = f"act=task_run_failed task={task.name} run_name={run_name} {format_exception(state)}"
    try:
        mario_log.warning(msg)
    except NameError:
        print(msg)

get_state_exception(state)

import inspect
from prefect.states import get_state_exception, aget_state_exception, format_exception
from prefect.exceptions import exception_traceback

def log_task_failure(task, task_run, state):
    run_name = getattr(task_run, "name", "(unknown)")
    exc = None
    try:
        exc = get_state_exception(state)
    except Exception:
        exc = None

    # 某些环境下会返回 coroutine；尝试桥接，否则回退
    if inspect.iscoroutine(exc):
        try:
            import anyio
            # 如果当前在工作线程里，这里能把协程交回事件循环执行
            exc = anyio.from_thread.run(aget_state_exception, state)
        except Exception:
            exc = None

    if isinstance(exc, BaseException):
        try:
            tb = exception_traceback(exc)
        except Exception:
            tb = ""
        msg = f"exc={exc}\n{tb}"
    else:
        # 拿不到 Exception（或不在可桥接的上下文）时，用官方格式化，包含完整堆栈
        msg = format_exception(state)

    try:
        mario_log.warning(f"act=task_run_failed task={task.name} run_name={run_name} {msg}")
    except NameError:
        print(f"act=task_run_failed task={task.name} run_name={run_name} {msg}")

format_exception(state)

prefect.exceptions.exception_traceback(exc)

get_state_exception(state)

import inspect
from prefect.states import get_state_exception, aget_state_exception, format_exception
from prefect.exceptions import exception_traceback

def _extract_exception_sync(state):
    """
    尝试同步地从 State 中提取 BaseException。
    如遇到 coroutine，使用 anyio.from_thread.run 桥接到当前事件循环。
    """
    try:
        exc = get_state_exception(state)
    except Exception:
        exc = None

    # 有些环境这里会是 coroutine，需要桥接
    if inspect.iscoroutine(exc):
        try:
            import anyio
            exc = anyio.from_thread.run(aget_state_exception, state)
        except Exception:
            exc = None

    return exc if isinstance(exc, BaseException) else None


def log_task_failure(task, task_run, state):
    run_name = getattr(task_run, "name", "(unknown)")
    exc = _extract_exception_sync(state)

    if exc is not None:
        # 完整 traceback（含行号/文件）
        try:
            tb = exception_traceback(exc)
        except Exception:
            tb = ""
        msg = f"exc={exc}\n{tb}"
    else:
        # 兜底：拿不到 Exception 时至少打印格式化信息
        # 如你的版本没有完整堆栈，这里可考虑升级 Prefect，或将 trace 放到任务内 try/except 里打日志
        msg = format_exception(state)

    try:
        mario_log.warning(f"act=task_run_failed task={task.name} run_name={run_name} {msg}")
    except NameError:
        print(f"act=task_run_failed task={task.name} run_name={run_name} {msg}")

exception_traceback(exc)

anyio.from_thread.run(...)

traceback.format_exc()

raise_state_exception(state)

import inspect
from prefect.states import get_state_exception, format_exception
try:
    from prefect.states import raise_state_exception  # 同步抛出状态里的异常
except ImportError:
    raise_state_exception = None
from prefect.exceptions import exception_traceback

def log_task_failure(task, task_run, state):
    run_name = getattr(task_run, "name", "(unknown)")

    exc = None
    # 尝试 1：直接拿异常（某些版本这里可能返回 coroutine，就跳过）
    try:
        e = get_state_exception(state)
        if isinstance(e, BaseException) and not inspect.iscoroutine(e):
            exc = e
    except Exception:
        pass

    # 尝试 2：用同步 API 把异常抛出来再捕获（拿到真实 Exception 和完整堆栈）
    if exc is None and callable(raise_state_exception):
        try:
            try:
                raise_state_exception(state)  # 这里会 raise 真实异常
            except BaseException as e:
                exc = e
        except Exception:
            pass

    if exc is not None:
        # 优先用 Prefect 的 traceback 格式化；失败则回退到 Python 标准库
        try:
            tb = exception_traceback(exc)
        except Exception:
            import traceback
            tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
        msg = f"exc={exc}\n{tb}"
    else:
        # 实在拿不到 Exception（或者版本不支持），兜底
        msg = format_exception(state)

    try:
        mario_log.warning(f"act=task_run_failed task={task.name} run_name={run_name} {msg}")
    except NameError:
        print(f"act=task_run_failed task={task.name} run_name={run_name} {msg}")

raise_state_exception(state)

get_state_exception

states

aget_state_exception

import inspect
import traceback as _traceback
from prefect.states import get_state_exception, raise_state_exception, format_exception
from prefect.exceptions import exception_traceback

def log_task_failure(_task_, _task_run_, _state_):
    run_name = getattr(_task_run_, "name", "(unknown)")

    exc = None
    # 尝试 1：直接取异常；部分环境可能返回 coroutine，就跳过
    try:
        e = get_state_exception(_state_)
        if isinstance(e, BaseException) and not inspect.iscoroutine(e):
            exc = e
    except Exception:
        pass

    # 尝试 2：同步抛出再捕获，拿到真实异常和完整堆栈（适配 async/sync）
    if exc is None:
        try:
            try:
                raise_state_exception(_state_)  # 这里会 raise 真实异常
            except BaseException as e:
                exc = e
        except Exception:
            exc = None

    if exc is not None:
        # 优先 Prefect 的 traceback；失败则回退到 Python 标准库
        try:
            tb = exception_traceback(exc)
        except Exception:
            tb = "".join(_traceback.format_exception(type(exc), exc, exc.__traceback__))
        msg = f"task={_task_.name} run_name={run_name} exc={exc}\n{tb}"
    else:
        # 实在拿不到 Exception（极少见），兜底
        msg = f"task={_task_.name} run_name={run_name} {format_exception(_state_)}"

    try:
        mario_log.warning(msg)
    except NameError:
        print(msg)

raise_state_exception(state)

import inspect
import traceback as _traceback

from prefect.states import get_state_exception, raise_state_exception
try:
    # 如果你的版本提供了异步版，优先用它
    from prefect.states import araise_state_exception
except Exception:
    araise_state_exception = None

try:
    from prefect.exceptions import exception_traceback
except Exception:
    exception_traceback = None


def _format_exc_with_tb(exc: BaseException) -> str:
    # 优先用 Prefect 的 traceback 格式化；没有就用标准库
    if exception_traceback:
        try:
            return exception_traceback(exc)
        except Exception:
            pass
    return "".join(_traceback.format_exception(type(exc), exc, exc.__traceback__))


def log_task_failure(_task_, _task_run_, _state_):
    run_name = getattr(_task_run_, "name", "(unknown)")

    exc = None
    # 尝试 1：直接拿异常；如果拿到的是 coroutine 就跳过
    try:
        e = get_state_exception(_state_)
        if isinstance(e, BaseException) and not inspect.iscoroutine(e):
            exc = e
    except Exception:
        pass

    # 尝试 2：让状态把异常“抛”出来，再在钩子里捕获（需要 anyio 桥接）
    if exc is None:
        try:
            import anyio

            if araise_state_exception is not None:
                # 异步抛异常：在同步钩子里通过 anyio.from_thread.run 执行
                try:
                    anyio.from_thread.run(araise_state_exception, _state_)
                except BaseException as e:
                    exc = e
            else:
                # 同步 API 可能返回 coroutine；判断并桥接执行
                res = None
                try:
                    res = raise_state_exception(_state_)  # 可能直接 raise，也可能返回 coroutine
                except BaseException as e:
                    exc = e

                if exc is None and inspect.iscoroutine(res):
                    async def _runner():
                        # 等待并让异常在这里抛出
                        await res

                    try:
                        anyio.from_thread.run(_runner)
                    except BaseException as e:
                        exc = e
        except Exception:
            # anyio 不可用或其他非关键失败，忽略
            pass

    if exc is not None:
        tb_text = _format_exc_with_tb(exc)
        msg = f"task={_task_.name} run_name={run_name} exc={exc}\n{tb_text}"
    else:
        # 实在拿不到 Exception（极少见），打印最基本的信息
        msg = f"task={_task_.name} run_name={run_name} (no exception object available)"

    try:
        mario_log.warning(msg)
    except NameError:
        print(msg)

raise_state_exception(state)

anyio.from_thread.run(...)

prefect.exceptions.exception_traceback(exc)

traceback.format_exception

araise_state_exception

from prefect import task, flow
from prefect.utilities.annotations import quote

@task
def do_something(server_record: quote):
    # server_record 会被按原样传入，不会被 Prefect 解析/遍历
    ...

@flow
def my_flow(server_record):
    do_something(server_record)  # 这里直接传即可

from prefect.utilities.annotations import quote

do_something(server_record=quote(server_record))

upload_to_bos

upload_data

quote

record_data = baikan_tasks.preproc_upload_to_bos(
    input_json, text_res, planning_tts_json, tts_res, ttv_res_list, video_path_list_url,
    cover_res, card_res_list, aigc_video_res_list, voice_effect_res, video_effect_res
)
server_record = upload_to_bos(record_data)

record_data

upload_to_bos

preproc_upload_to_bos

tolist()

to_dict(orient="records")

to_json()

@task
def preproc_upload_to_bos(...):
    arr = np.arange(3).tolist()
    df = some_df.to_dict(orient="records")
    return {"arr": arr, "df": df, "other": "ok"}

from prefect.utilities.annotations import quote
from prefect import task

@task
def preproc_upload_to_bos(...):
    # 假设 df / binary_data / custom_obj 是解析时会出问题的值
    record_data = {
        "meta": {...},
        "df": quote(df),                  # 跳过递归解析 df
        "binary": quote(binary_data),     # 比如 bytes
        "custom": quote(custom_obj),      # 自定义对象
    }
    return record_data

@task
def upload_to_bos(upload_data):
    # 这里会收到完整的 dict，但 df/binary/custom 不会被 Prefect 递归展开
    ...

preproc_upload_to_bos

from prefect import get_run_logger

@task
def preproc_upload_to_bos(...):
    record_data = {...}
    logger = get_run_logger()
    <http://logger.info|logger.info>({k: type(v).__name__ for k, v in record_data.items()})
    return record_data

record_data

quote(record_data)

record_data

upload_to_bos(upload_data=record_data)

upload_data

preproc_upload_to_bos

ttv_res/card_res_list/aigc_video_res_list

record_data

upload_to_bos

preproc_upload_to_bos

record_data

from prefect import get_run_logger

def _walk_types(x, path=""):
    t = type(x).__name__
    if isinstance(x, dict):
        return {k: _walk_types(v, f"{path}.{k}") for k, v in x.items()}
    elif isinstance(x, (list, tuple)):
        return [ _walk_types(v, f"{path}[{i}]") for i, v in enumerate(x) ]
    else:
        return t

@task
def preproc_upload_to_bos(...):
    record_data = {...}  # 你构造的返回
    logger = get_run_logger()
    <http://logger.info|logger.info>(_walk_types(record_data))
    return record_data

tolist()

to_dict(orient="records")

to_json()

from prefect.utilities.annotations import quote

@task
def preproc_upload_to_bos(...):
    record_data = {
        "ok_meta": {...},
        "maybe_binary": quote(binary_data),   # 避免 Prefect 递归进去
        "custom_obj": quote(custom_obj),      # 自定义对象也可 quote
    }
    return record_data

record_data

record_data

json.dumps

preproc_upload_to_bos

PrefectFutureList

preproc_upload_to_bos

record_data

upload_data

record_data

upload_to_bos(upload_data=record_data)

upload_data

from prefect import task

@task
def preproc_upload_to_bos(
    input_json, text_res, planning_tts_json, tts_res, ttv_res, video_path_list_url,
    cover_res, card_res_list, aigc_video_res_list, voice_effect_res, video_effect_res
):
    # 这里使用的都是“参数本身”，Prefect 在进入 task 前会先把 futures 解析为真实值
    ret =  {
        "input_json": input_json,
        "text_res": text_res,
        "planning_tts_json": planning_tts_json,
        "tts_res": tts_res,
        "ttv_res": ttv_res,  # 应该是普通 list/数据结构，不再是 PrefectFutureList
        "video_path_list_url": video_path_list_url,
        "cover_res": cover_res,
        "card_res_list": card_res_list,
        "aigc_video_res_list": aigc_video_res_list,
        "voice_effect_res": voice_effect_res,
        "video_effect_res": video_effect_res,
    }

    # 打印叶子类型，确认都已解析为基础类型
    ret_type = {k: type(v).__name__ for k, v in ret.items()}
    logger = get_run_logger()
    <http://logger.info|logger.info>(ret_type)

    # 如果 fs.save 需要字符串，记得序列化
    # fs.save(_query_=input_json["task_id"], _file_name_="preproc_upload_to_bos", _text_=json.dumps(ret, ensure_ascii=False))
    # fs.save(_query_=input_json["task_id"], _file_name_="preproc_upload_to_bos_type", _text_=json.dumps(ret_type, ensure_ascii=False))

    return ret

ret

quote

ret

from prefect.utilities.annotations import quote

@task
def preproc_upload_to_bos(...):
    ...
    ret = {
        "input_json": input_json,
        "df": quote(df),                 # pandas DataFrame -> 跳过 Prefect 递归
        "binary": quote(binary_bytes),   # 二进制内容
        "custom": quote(custom_obj),     # 自定义对象
        ...
    }
    return ret

ret_type

upload_to_bos(upload_data=record_data)

upload_data

@flow

PrefectFutureList

from prefect import flow

@flow
def main():
    record_data = preproc_upload_to_bos(...所有上游结果...)
    server_record = upload_to_bos(record_data)

if __name__ == "__main__":
    main()

from prefect.futures import resolve_futures_to_results
from prefect import task

@task
def preproc_upload_to_bos(...):
    ttv_res = resolve_futures_to_results(ttv_res)
    card_res_list = resolve_futures_to_results(card_res_list)
    aigc_video_res_list = resolve_futures_to_results(aigc_video_res_list)

    ret = {
        "input_json": input_json,
        "text_res": text_res,
        "planning_tts_json": planning_tts_json,
        "tts_res": tts_res,
        "ttv_res": ttv_res,  # 现在是普通 list
        "video_path_list_url": video_path_list_url,
        "cover_res": cover_res,
        "card_res_list": card_res_list,  # 普通 list
        "aigc_video_res_list": aigc_video_res_list,  # 普通 list
        "voice_effect_res": voice_effect_res,
        "video_effect_res": video_effect_res
    }
    return ret

quote

ret

from prefect.utilities.annotations import quote

ret = {
  ...
  "binary": quote(binary_bytes),
  "custom": quote(custom_obj),
}

quote

fs.save(..., text=ret)

fs.save

json.dumps(ret, ensure_ascii=False)

@flow

log

timestamp

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_timestamp_brin ON log USING brin (timestamp) WITH (pages_per_range = 32);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_flow_run_id ON log (flow_run_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_log_task_run_id ON log (task_run_id);

-- 每次删 100k，走 timestamp 范围扫描（需要前面的索引）
WITH batch AS (
  SELECT id
  FROM log
  WHERE timestamp < now() - interval '12 days'
  ORDER BY timestamp
  LIMIT 100000
)
DELETE FROM log l
USING batch
WHERE l.id = batch.id;

log

log

log

timestamp

timestamp

log_old

ALTER TABLE log DETACH PARTITION log_2025_10; DROP TABLE log_2025_10;

log

-- 0) 日志按时间分批清理（先执行）
WITH batch AS (
  SELECT id
  FROM log
  WHERE timestamp < now() - interval '12 days'
  ORDER BY timestamp
  LIMIT 100000
)
DELETE FROM log l
USING batch
WHERE l.id = batch.id;
-- 循环上述块直到 0 行

log.timestamp

now()

log

PREFECT_LOGGING_TO_API_ENABLED

log

FlowRunContext.get()

from prefect import flow, task
from prefect.context import FlowRunContext

def log_task_failure(task, task_run, state):
    ctx = FlowRunContext.get()
    if ctx and ctx.flow_run:
        print(f"Flow run name: {ctx.flow_run.name}")
        print(f"Flow run id: {ctx.flow_run.id}")
    # 任务本身的信息也在参数里
    print(f"Task run name: {task_run.name}")
    print(f"Task run id: {task_run.id}")

@task(on_failure=[log_task_failure])
def my_task():
    raise RuntimeError("boom")

@flow
def my_flow():
    my_task()

if __name__ == "__main__":
    my_flow()

task_run.flow_run_id

from prefect.client.orchestration import get_client

async def log_task_failure_async(task, task_run, state):
    async with get_client() as client:
        flow_run = await client.read_flow_run(task_run.flow_run_id)
        print(f"Flow run name: {flow_run.name}")

from prefect import flow, task
from prefect.futures import wait
from prefect.states import get_state_exception
from prefect.context import FlowRunContext

@task
def task_a(x): ...
@task
def task_b(x): ...
@task
def task_c(x): ...
@task
def task_d(tasks: dict): ...

@flow(flow_run_name="{logid}")
def my_flow(input_data, logid):
    ctx = FlowRunContext.get()
    flow_run_id = ctx.flow_run.id

    # 提前用 dict 记录每个 future，方便失败时识别是哪个任务
    tasks = {}

    try:
        a = task_a.submit(input_data)
        tasks['a'] = a

        b = task_b.with_options(retries=3).submit(a)
        c = task_c.submit(a)
        tasks['b'] = b
        tasks['c'] = c

        if a.result(raise_on_failure=True)["value"] > 30:
            e = task_e.submit(a)
            tasks['e'] = e

        d = task_d.submit(tasks)
        tasks['d'] = d

        # 等全部完成，再统一检查状态（也可以按需早检查）
        done, _ = wait(list(tasks.values()))
        failed = {}
        for name, fut in tasks.items():
            if fut.state.is_failed() or fut.state.type.name in ("CANCELLED", "CRASHED"):
                exc = get_state_exception(fut.state)  # 不会抛出，直接取异常对象
                failed[name] = {
                    "task_run_id": str(fut.task_run_id),
                    "state": fut.state.type.name,
                    "message": fut.state.message,
                    "exception": str(exc) if exc else None,
                }

        if failed:
            return {
                "flow_run_id": str(flow_run_id),
                "errno": 1,
                "errmsg": "some tasks failed",
                "failed": failed,
            }

        return {
            "flow_run_id": str(flow_run_id),
            "errno": 0,
            "errmsg": "",
            "result": d.result(raise_on_failure=True),
        }

    except Exception as e:
        # try/except 里也可以把已提交的任务做一次状态汇总
        done, _ = wait(list(tasks.values())) if tasks else (set(), set())
        failed = {}
        for name, fut in tasks.items():
            if fut.state.is_failed() or fut.state.type.name in ("CANCELLED", "CRASHED"):
                exc = get_state_exception(fut.state)
                failed[name] = {
                    "task_run_id": str(fut.task_run_id),
                    "state": fut.state.type.name,
                    "message": fut.state.message,
                    "exception": str(exc) if exc else None,
                }
        return {
            "flow_run_id": str(flow_run_id),
            "errno": 1,
            "errmsg": str(e),
            "failed": failed,
        }

tasks = {'b': b, 'c': c, ...}

future.state.is_failed()

future.state.type

get_state_exception(future.state)

future.result(raise_on_failure=True)

CANCELLED

task_a/b/c/d/e

get_state()

future.state

wait()

from prefect.futures import PrefectFuture, PrefectFutureList, wait
from prefect.client.schemas.objects import StateType

failed_tasks_name = []

# 先收集所有 future，统一等它们完成
local_vars = locals()
all_futures = []
for var_val in local_vars.values():
    if isinstance(var_val, PrefectFuture):
        all_futures.append(var_val)
    elif isinstance(var_val, (list, PrefectFutureList)):
        all_futures.extend([f for f in var_val if isinstance(f, PrefectFuture)])

if all_futures:
    wait(all_futures)  # 等所有任务完成，避免状态还在 RUNNING/PENDING

# 再判断失败的变量名
for var_name, var_val in local_vars.items():
    if isinstance(var_val, PrefectFuture):
        st = var_val.state
        if st.is_failed() or st.type in {StateType.CRASHED, StateType.CANCELLED}:
            failed_tasks_name.append(var_name)

    elif isinstance(var_val, (list, PrefectFutureList)):
        if not var_val:
            continue
        # 只要列表里有一个失败/崩溃/取消，就把这个变量名记为失败集合
        if any(
            isinstance(fut, PrefectFuture) and
            (fut.state.is_failed() or fut.state.type in {StateType.CRASHED, StateType.CANCELLED})
            for fut in var_val
        ):
            failed_tasks_name.append(var_name)

tasks = {}
tasks['b'] = task_b.with_options(retries=3).submit(a)
tasks['c'] = task_c.submit(a)
...
done, _ = wait(list(tasks.values()))
failed = [name for name, fut in tasks.items()
          if fut.state.is_failed() or fut.state.type in {StateType.CRASHED, StateType.CANCELLED}]

future.task_run_id

TaskRun

my_flow

# 假设 fut 是 PrefectFuture
val = fut.result(raise_on_failure=False)  # 失败时返回异常对象，不会抛错
if isinstance(val, Exception):
    concise = str(val)              # -> "blade_req_final_failed"
    # 也可更保险：
    if not concise and val.args:
        concise = str(val.args[0])

from prefect.states import get_state_exception

st = fut.state
if st.is_failed() or st.is_crashed():
    exc = get_state_exception(st)   # 不会抛错，返回 Exception 或 None
    concise = str(exc) if exc else (st.message or "")

from prefect.states import get_state_exception

def on_task_failure(task, task_run, state):
    exc = get_state_exception(state)
    concise = str(exc) if exc else (state.message or "")
    print(concise)  # "blade_req_final_failed"

str(exc)

RuntimeError("blade_req_final_failed")

state.message

concise = state.message.split(":", 1)[-1].strip() if state.message else ""

str(exc)

raise_on_failure=False

str()

task_run_id

var_val.state

task_name

from prefect.futures import PrefectFuture, PrefectFutureList, wait
from prefect.client.orchestration import get_client
from prefect.client.schemas.objects import StateType
from prefect.states import get_state_exception

def get_failed_tasks(local_vars):
    """返回 (failed_task_run_name, concise_error_msg)，如未发现失败则为空字符串"""
    failed_task_name = ""
    failed_task_msg = ""

    # 可选：如果需要，先等待这些 futures 完成，避免读取到 RUNNING/PENDING
    all_futs = []
    for v in local_vars.values():
        if isinstance(v, PrefectFuture):
            all_futs.append(v)
        elif isinstance(v, (list, PrefectFutureList)):
            all_futs.extend([f for f in v if isinstance(f, PrefectFuture)])
    if all_futs:
        wait(all_futs)

    def concise_err(st):
        exc = get_state_exception(st)
        return str(exc) if exc else (st.message or "")

    with get_client(sync_client=True) as client:
        def check_future(fut):
            nonlocal failed_task_name, failed_task_msg
            st = fut.state
            if st.is_failed() or st.type in {StateType.CRASHED, StateType.CANCELLED}:
                tr = client.read_task_run(fut.task_run_id)
                failed_task_name = tr.name             # 任务运行名
                failed_task_msg = concise_err(st)      # 精简错误
                return True
            return False

        for var_val in local_vars.values():
            if isinstance(var_val, PrefectFuture):
                if check_future(var_val):
                    break
            elif isinstance(var_val, (list, PrefectFutureList)):
                for fut in var_val:
                    if isinstance(fut, PrefectFuture) and check_future(fut):
                        break
                if failed_task_name:
                    break

    return failed_task_name, failed_task_msg

from prefect.futures import wait
from prefect.client.schemas.objects import StateType
from prefect.states import get_state_exception

def collect_failed_from_mapping(tasks: dict[str, PrefectFuture]):
    # tasks 形如：{"task_b": b_future, "task_c": c_future, ...}
    if tasks:
        wait(list(tasks.values()))
    for name, fut in tasks.items():
        st = fut.state
        if st.is_failed() or st.type in {StateType.CRASHED, StateType.CANCELLED}:
            msg = str(get_state_exception(st) or "") or (st.message or "")
            return name, msg
    return "", ""

tasks = {}
tasks["task_b"] = task_b.with_options(retries=3).submit(a)
tasks["task_c"] = task_c.submit(a)
# 如果有列表任务，建议用 "task_x[0]" 这类标记来存 key
failed_name, failed_msg = collect_failed_from_mapping(tasks)

failed_task_msg = get_err_msg(var_val.state)

task_obj.name

get_client

task_obj.name

from prefect.futures import PrefectFuture, PrefectFutureList, wait
from prefect.client.schemas.objects import StateType
from prefect.states import get_state_exception

def get_failed_tasks_from_map(tasks_map: dict[str, PrefectFuture] | None,
                              lists_map: dict[str, list[PrefectFuture]] | None = None):
    """
    tasks_map 例子: {"task_b": b_future, "task_c": c_future}
    lists_map 例子: {"task_x": [x_fut0, x_fut1, ...]}
    返回 (failed_task_name, concise_error_msg)
    """
    tasks_map = tasks_map or {}
    lists_map = lists_map or {}

    # 等待所有提交完成（避免读到 RUNNING/PENDING）
    all_futs = list(tasks_map.values()) + [f for lst in lists_map.values() for f in lst]
    if all_futs:
        wait(all_futs)

    def concise_err(st):
        exc = get_state_exception(st)
        return str(exc) if exc else (st.message or "")

    # 先查单个任务
    for name, fut in tasks_map.items():
        st = fut.state
        if st.is_failed() or st.type in {StateType.CRASHED, StateType.CANCELLED}:
            return name, concise_err(st)

    # 再查列表任务（任一元素失败就返回该列表名）
    for name, futs in lists_map.items():
        for fut in futs:
            st = fut.state
            if st.is_failed() or st.type in {StateType.CRASHED, StateType.CANCELLED}:
                return name, concise_err(st)

    return "", ""

tasks_map = {}
tasks_map[task_b.name] = task_b.with_options(retries=3).submit(a)
tasks_map[task_c.name] = task_c.submit(a)

lists_map = {}
# 如果有批量任务，给它一个集合名
# lists_map["task_x"] = [task_x.submit(i) for i in items]

failed_name, failed_msg = get_failed_tasks_from_map(tasks_map, lists_map)

ObjectNotFound

var_val.state

my_flow

setdefault

name

task_run_name

from prefect import task as prefect_task

def task(__fn=None, **kwargs):
    # 统一默认
    kwargs.setdefault("cache_result_in_memory", False)
    kwargs.setdefault("persist_result", False)
    kwargs.setdefault("cache_policy", NO_CACHE)
    kwargs.setdefault("on_failure", [log_task_failure])

    def _decorate(fn):
        # 默认让 Task 定义名 与 Task Run 名称 都等于函数名
        kwargs.setdefault("name", fn.__name__)
        kwargs.setdefault("task_run_name", fn.__name__)
        return prefect_task(**kwargs)(fn)

    # 支持 @task 和 @task(...)
    return _decorate if __fn is None else _decorate(__fn)

name

with_options(task_run_name=...)

task_run_name

kwargs.setdefault(
      "task_run_name",
      lambda ctx, params: fn.__name__  # 这里可以根据 params 组装更丰富的名字
  )

TaskRun

task_run.task_key

def extract_func_name_from_task_key(task_key: str) -> str:
    # task_key 形如 "my_task-a1b2c3d4" 或 "package.module.MyClass.my_task-a1b2c3d4"
    # 先去掉模块/类限定，再去掉哈希
    return task_key.split(".")[-1].split("-")[0]

from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterFlowRunId

task_runs = await client.read_task_runs(
    task_run_filter=TaskRunFilter(
        flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run.id])
    )
)

func_names = [extract_func_name_from_task_key(tr.task_key) for tr in task_runs]

task_run.name

task_run_name

task_run.name

extract_func_name_from_task_key

state

from prefect.futures import PrefectFuture, PrefectFutureList, wait
from prefect.client.schemas.objects import StateType

def get_failed_tasks(local_vars, timeout=None):
    print("True Order:", list(local_vars.keys()))

    # 先收集并等待
    all_futs = []
    for v in local_vars.values():
        if isinstance(v, PrefectFuture):
            all_futs.append(v)
        elif isinstance(v, PrefectFutureList):
            all_futs.extend([f for f in v if isinstance(f, PrefectFuture)])
    done, not_done = wait(all_futs, timeout=timeout) if all_futs else (set(), set())

    def is_failed(st):
        return st.is_failed() or st.type in {StateType.CRASHED, StateType.CANCELLED}

    failed_var = ""
    failed_msg = ""

    for var_name, var_val in local_vars.items():
        if isinstance(var_val, PrefectFuture):
            st = var_val.state
            if is_failed(st):
                failed_var = var_name
                failed_msg = get_err_msg(st)
                break

        elif isinstance(var_val, PrefectFutureList):
            if not var_val:
                continue
            for fut in var_val:
                if not isinstance(fut, PrefectFuture):
                    continue
                st = fut.state
                if is_failed(st):
                    failed_var = var_name
                    failed_msg = get_err_msg(st)
                    break
            if failed_var:
                break

    # 可选：如果有未完成任务，按需视为超时/失败
    if not failed_var and not_done:
        failed_var = "pending_tasks"
        failed_msg = f"{len(not_done)} task(s) not finished"
    return failed_var, failed_msg

val = fut.result(raise_on_failure=False)  # 等到完成，不会抛异常；失败时返回异常对象
st = fut.state  # 此时为终态

for fut in var_val:
    fut.result(raise_on_failure=False)
    if fut.state.is_failed():
        ...

from prefect.futures import as_completed
from prefect.client.schemas.objects import StateType

def get_first_failed(futs):
    for fut in as_completed(futs, timeout=None):
        st = fut.state
        if st.is_failed() or st.type in {StateType.CRASHED, StateType.CANCELLED}:
            return fut
    return None

tasks = {}
  tasks["task_b"] = task_b.submit(...)
  tasks["task_c"] = task_c.submit(...)
  done, _ = wait(list(tasks.values()))
  failed = [name for name, fut in tasks.items() if fut.state.is_failed()]

from prefect.states import get_state_exception

prefect.futures.wait()

wait()

wait()

future.state

.state

# 反例：会造成高频 API 调用
    while not fut.state.is_final():
        time.sleep(0.1)

wait

timeout=60

.state

from prefect.futures import wait, resolve_futures_to_states

  def get_failed_tasks(local_vars):
      failed = []

      # 收集 PrefectFuture
      futs = []
      for v in local_vars.values():
          if isinstance(v, PrefectFuture):
              futs.append(v)
          elif isinstance(v, PrefectFutureList):
              futs.extend([f for f in v if isinstance(f, PrefectFuture)])

      # 一次性等待（事件驱动，无线程暴增）
      if futs:
          wait(futs, timeout=60)

      # 批量解析状态（避免逐个触发多次 API）
      states = resolve_futures_to_states(futs) if futs else []

      for fut, st in zip(futs, states):
          if st.is_failed():
              # 按需获取错误信息：如果你有自定义的 get_err_msg(st) 可以放这里
              failed.append((fut, st))

      return failed

as_completed

wait