https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> If I want to create an automation tha...
# ask-marvin
t
@Marvin If I want to create an automation that triggers a flow run when there's no more certain flow in 
Late
or 
Scheduled
, how would I write the automation? I am confortable with using python API instead of the Web UI.
m
✅ thought for 755.0 seconds
Great question. Quick clarifier: are you on Prefect Cloud 3.x? Automations are a Cloud feature. If you’re running open-source Server, the closest equivalent is a small “watchdog” flow that polls and triggers the next run. If Cloud is available, the intent with Automations would be a metric trigger that evaluates “count of flow runs for FLOW_X in states Scheduled or Late equals 0 (optionally for N seconds)” and performs a Run Deployment action. I can draft the exact Automation Python if you confirm: - the source flow name you want to watch - the deployment you want to run when the queue is empty - whether you want a stabilization window (e.g., require the condition to hold for ~60s) In the meantime, here’s a reliable, fully-Python alternative you can use right now (works in both Cloud and OSS). It checks for any Scheduled or Late runs of a given flow and, if none are found, triggers a deployment:
Copy code
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowFilter, FlowFilterName,
    FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType,
)
from prefect.client.schemas.objects import StateType
from prefect.deployments.flow_runs import run_deployment

@flow(name="trigger-when-no-pending")
async def trigger_when_no_scheduled_or_late(
    target_flow_name: str,        # e.g., "upstream-flow"
    deployment_to_run: str,       # e.g., "downstream-flow/downstream-deployment"
    run_tags: list[str] | None = None
):
    async with get_client() as client:
        # Filter flow runs for the target flow
        flow_filter = FlowFilter(name=FlowFilterName(any_=[target_flow_name]))

        # Filter by state types Scheduled or Late
        state_type_filter = FlowRunFilterStateType(any_=[StateType.SCHEDULED, StateType.LATE])
        state_filter = FlowRunFilterState(type=state_type_filter)
        flow_run_filter = FlowRunFilter(state=state_filter)

        # Ask for just one matching run to know if any exist
        pending = await client.read_flow_runs(
            flow_filter=flow_filter,
            flow_run_filter=flow_run_filter,
            limit=1,
        )

        # If none are pending, trigger the deployment
        if not pending:
            await run_deployment(
                name=deployment_to_run,
                tags=run_tags
            )

if __name__ == "__main__":
    # Example invocation
    trigger_when_no_scheduled_or_late(
        target_flow_name="upstream-flow",
        deployment_to_run="downstream-flow/downstream-deployment",
        run_tags=["auto-triggered"]
    )
How to use it: - Deploy this “watchdog” flow with an Interval or Cron schedule (e.g., every 1–2 minutes). - It will run, detect if there are any Scheduled or Late runs for 
target_flow_name
, and if not, kick off 
deployment_to_run
. If you confirm you’re on Cloud and share the exact flow/deployment names, I’ll translate this into a Cloud Automation with a Metric trigger that performs the same logic (i.e., “count of Scheduled/Late = 0 for N seconds” => Run Deployment) so you don’t need a polling flow.
t
@Marvin I am self-hosting prefect server, but I have automation set up and it works?
m
✅ thought for 44.7 seconds
Got it — thanks for clarifying. In recent Prefect Server 3.x stacks, the Events/Automations services can be enabled, so it’s possible to run automations self‑hosted. That’s likely what you’re seeing. The CLI doesn’t expose a “create” command, but you can still create them via the UI or by calling the API directly. Two solid paths forward: Option A — Reliable now (works in Server and Cloud): a small “watchdog” flow - This checks if a given flow has any runs in Scheduled or Late and, if none are found, triggers your downstream deployment. - You can run it on an Interval/Cron. Code:
Copy code
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowFilter, FlowFilterName,
    FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType,
)
from prefect.client.schemas.objects import StateType
from prefect.deployments.flow_runs import run_deployment

@flow(name="trigger-when-no-pending")
async def trigger_when_no_scheduled_or_late(
    target_flow_name: str,        # e.g., "upstream-flow"
    deployment_to_run: str,       # e.g., "downstream-flow/downstream-deployment"
    run_tags: list[str] | None = None
):
    async with get_client() as client:
        flow_filter = FlowFilter(name=FlowFilterName(any_=[target_flow_name]))
        state_type_filter = FlowRunFilterStateType(any_=[StateType.SCHEDULED, StateType.LATE])
        state_filter = FlowRunFilterState(type=state_type_filter)
        flow_run_filter = FlowRunFilter(state=state_filter)

        pending = await client.read_flow_runs(
            flow_filter=flow_filter,
            flow_run_filter=flow_run_filter,
            limit=1,
        )

        if not pending:
            await run_deployment(
                name=deployment_to_run,
                tags=run_tags
            )
Option B — Create an Automation via the Server API (Python) - This is possible, but the exact JSON shape for a MetricTrigger depends on your Server version’s schema. - The general idea is: create an Automation with a Metric trigger “count of flow runs for FLOW_X in states [Scheduled, Late] equals 0 (optionally for N seconds)” and a RunDeployment action. If you want me to generate a ready-to-run Python snippet that POSTs the correct payload, please share: - Your Prefect Server version (prefect version) - The flow name you want to monitor - The deployment name to run - Any desired stabilization window (e.g., 60 seconds) Alternatively, you can pull the exact schema from your server: - Open your API’s OpenAPI docs (e.g., your API URL’s OpenAPI JSON) and look for the “AutomationCreate” and “MetricTrigger” schemas. I can translate that into a Python `httpx`/`requests` call for you. Let me know which route you prefer. If you want pure Automations, I’ll draft the precise request for your server version. If you want something you can deploy immediately, Option A will do the job.
t
@Marvin I am on the latest 3.4.22 so I want automation. More specifically, I want it such that when there's no 
inference/inference
deployment in Late and Schedule in 10 minutes, run 
add-inference-job-to-queue/add-inference-job-to-queue
flow with default parameters.
m
✅ thought for 223.9 seconds
t
@Marvin 用中文回答一遍：如果我想要在Python里面创造这个Automation，我应该怎么做？
m
✅ thought for 77.7 seconds
可以的！你在自建的 Prefect Server 3.x 上启用了 Events/Automations 服务，所以可以用 Python 创建 Automation。不过不同版本的 Automation schema（尤其是 MetricTrigger 部分）有些细微差异。下面给你两种做法： 方案 A（立刻可用，最稳妥）：用“看门狗”Flow轮询判断，没有 Scheduled/Late 就触发下游 - 这个做法不依赖 Automation 的 schema，Server / Cloud 都能跑 - 将此 Flow 部署为一个按间隔运行的部署（例如每 1-2 分钟） 代码示例：
Copy code
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowFilter, FlowFilterName,
    FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType,
)
from prefect.client.schemas.objects import StateType
from prefect.deployments.flow_runs import run_deployment

@flow(name="trigger-when-no-pending")
async def trigger_when_no_scheduled_or_late(
    target_flow_name: str,        # 需要监控的上游 Flow 名称
    deployment_to_run: str,       # 没有待运行时要触发的 Deployment（例如 "downstream-flow/downstream-deployment"）
    run_tags: list[str] | None = None
):
    async with get_client() as client:
        # 过滤目标 Flow
        flow_filter = FlowFilter(name=FlowFilterName(any_=[target_flow_name]))
        # 过滤状态为 Scheduled 或 Late 的 Flow Run
        state_type_filter = FlowRunFilterStateType(any_=[StateType.SCHEDULED, StateType.LATE])
        state_filter = FlowRunFilterState(type=state_type_filter)
        flow_run_filter = FlowRunFilter(state=state_filter)

        # 查 1 条看看是否存在
        pending = await client.read_flow_runs(
            flow_filter=flow_filter,
            flow_run_filter=flow_run_filter,
            limit=1,
        )

        # 如果没有任何 Scheduled/Late，触发下游部署
        if not pending:
            await run_deployment(
                name=deployment_to_run,
                tags=run_tags
            )

if __name__ == "__main__":
    # 示例调用
    trigger_when_no_scheduled_or_late(
        target_flow_name="upstream-flow",
        deployment_to_run="downstream-flow/downstream-deployment",
        run_tags=["auto-triggered"]
    )
方案 B（用 Python 直接创建 Automation） - 思路：创建一个 MetricTrigger，表达“某个 Flow 的 Scheduled/Late 运行数 == 0（可选：持续 N 秒）”，动作使用 RunDeployment - 需要根据你的 Server 所在版本的 schema 构造正确的字段（MetricTrigger 的字段命名在不同小版本略有变化） 大致代码骨架（请把占位处替换为你的实际字段；如遇校验错误，把 prefect 版本发我，我按你的版本给你改成可运行的成品）： ``` from prefect.client.orchestration import get_client # 这些类在你当前安装版本中可能位于 prefect.automations 或 prefect.events.schemas.automations / prefect.events.actions 下 from prefect.automations import ( AutomationCore, Posture, MetricTrigger, MetricTriggerQuery, PrefectMetric, MetricTriggerOperator, RunDeployment, ) async def create_no_pending_automation( automation_name: str, target_flow_name: str, # 需要监控的 Flow 名称 deployment_to_run: str, # "flow_name/deployment_name" window_seconds: int = 60, # 条件需稳定维持的时间窗口（可选） enabled: bool = True, ): # 下面这段 MetricTriggerQuery 的具体字段名，可能因版本不同而略有差异 # 比如 metric、filters、operator、value/threshold、window/window_seconds 等 trigger = MetricTrigger( posture=Posture.Reactive, # 或 Posture.Proactive（取决于你的需求/版本） query=MetricTriggerQuery( # 举例：统计目标 flow 在 Scheduled/Late 状态下的 run 数 # metric=PrefectMetric.flow_runs, # 具体枚举名称/写法请以你的版本为准 # filters={ # "flow": {"name": {"any_": [target_flow_name]}}, # "flow_run": {"state": {"type": {"any_": ["SCHEDULED", "LATE"]}}}, # }, # operator=MetricTriggerOperator.EQ, # value=0, # window_seconds=window_seconds, ), ) action = RunDeployment( # 一般支持通过名字或 ID 指定部署；字段名在不同版本可能为 deployment_name / deployment_id # deployment_name=deployment_to_run, ) automation = AutomationCore( name=automation_name, description=f"当 {target_flow_name} 不再有 Scheduled/Late 时触发 {deployment_to_run}", enabled=enabled, trigger=trigger, actions=[action], ) async with get_client() as client: # PrefectClient 暴露了 create_automation；参数类型通常是 AutomationCreate（AutomationCore 兼容）
created = await client.create_automation(automation=automation) return created # 使用时： # await create_no_pending_automation( # automation_name="no-pending-then-run-downstream", # target_flow_name="upstream-flow", # deployment_to_run="downstream-flow/downstream-deployment", # window_seconds=60, # ) ``` 说明与建议： - Prefect CLI 暂无 
prefect automations create
子命令，这是正常的；用 UI 或 Python/API 即可。 - MetricTrigger 的字段（例如 metric 名、过滤器写法、window 字段名）在不同 3.x 小版本有微调。如果你把 `prefect version`、目标 Flow 名、要触发的 Deployment 名告诉我，我可以直接给出与你版本完全匹配、可直接运行的 Python 脚本。 - 如果你要马上落地，先用“方案 A 看门狗 Flow”最稳；之后再把逻辑迁到 Automation 也不迟。 需要我根据你的具体版本把“方案 B”的代码补全成可执行版本吗？
t
bro is mr. worldwise
@Marvin 我在使用3.4.22，帮我把方案B的代码补全
m
✅ thought for 69.6 seconds
Thanks — 3.4.22 Server can run Automations, so we can do this with a Metric trigger. Goal restated: - When there are no runs for deployment inference/inference in states Scheduled or Late for 10 minutes, run deployment add-inference-job-to-queue/add-inference-job-to-queue with default params. Here’s a Python snippet that creates such an automation by calling the Server API directly. It uses a metric “flow run count” filtered to that deployment and state types, and requires the condition to hold for 600 seconds before firing the action. Notes: - The metric and field names are server-schema specific; in 3.4.22 they typically follow the pattern shown below. - If anything 400s, run prefect automations inspect <automation_id> on a similar UI-made automation to confirm the exact JSON shape and I’ll adjust quickly. Code:
Copy code
import asyncio
from prefect.client.orchestration import get_client

# Update these if your names differ
SOURCE_FLOW_NAME = "inference"
SOURCE_DEPLOYMENT_NAME = "inference"  # the deployment for the inference flow
TARGET_DEPLOYMENT_SLUG = "add-inference-job-to-queue/add-inference-job-to-queue"  # flow/deployment

async def create_automation():
    async with get_client() as client:
        # Automation payload
        payload = {
            "name": "Run add-inference when inference has no pending runs for 10m",
            "description": "If inference/inference has no Scheduled or Late runs for 10 minutes, run add-inference-job-to-queue.",
            "enabled": True,
            "posture": "Reactive",
            "trigger": {
                "type": "metric",
                # The condition must hold for 600s before firing
                "firing_for_seconds": 600,
                "queries": [
                    {
                        # Count of flow runs
                        "metric": "prefect.flow_run.count",
                        # equals 0
                        "operator": "==",
                        "value": 0,
                        # Filter just the runs we care about
                        "where": {
                            # constrain to the specific flow and deployment
                            "flow.name": {"any_": [SOURCE_FLOW_NAME]},
                            "deployment.name": {"any_": [SOURCE_DEPLOYMENT_NAME]},
                            # only consider these state types
                            "state.type": {"any_": ["SCHEDULED", "LATE"]},
                        },
                        # Optionally, a rolling lookback window; can omit if not required
                        # "range_seconds": 600,
                    }
                ],
            },
            "actions": [
                {
                    "type": "run-deployment",
                    # You can pass the slug; server will resolve it
                    "deployment": TARGET_DEPLOYMENT_SLUG,
                    # Omit parameters to use defaults
                    # "parameters": {},
                    # Optional: tags for the triggered run
                    # "tags": ["auto-triggered"],
                }
            ],
        }

        # Create the automation
        resp = await <http://client._client.post|client._client.post>("/automations/", json=payload)
        resp.raise_for_status()
        print("Created automation:", resp.json()["id"])

if __name__ == "__main__":
    asyncio.run(create_automation())
A couple of tips: - The “firing_for_seconds: 600” ensures there’s a continuous 10-minute period with zero Scheduled/Late runs before the action fires; any new Scheduled/Late run during that time resets the timer. - If your server uses slightly different field names for metrics or filters, create this automation once in the UI with the same settings, then run: - 
prefect automations ls
to find its ID - 
prefect automations inspect <ID>
to see the exact JSON. Mirror that JSON in the code above.
- For reference on Automations types, see the source at prefect/automations.py. If you want, I can quickly validate the exact JSON for your server: share the output of 
prefect automations inspect <ID>
from a similar UI-created automation, and I’ll adapt the code precisely.
没问题！你在 Prefect Server 3.4.22 上可以用 Python 直接创建 Automation。下面给你一份“当某个 Flow 没有任何 Scheduled/Late 的运行时，自动触发某个部署”的完整示例。注意：MetricTrigger 在不同 3.x 小版本的字段可能有细微差异（主要是查询字段名），我已按 3.4.x 常见用法写好，并在注释中给出如果报校验错误的调整方式。 代码（直接可用，按注释替换你的 Flow/Deployment 名称）：
Copy code
import asyncio
from prefect.client.orchestration import get_client

# 这组类在 3.4.x 中通常可以从 prefect.automations 导入
from prefect.automations import (
    AutomationCore,
    MetricTrigger,
    MetricTriggerQuery,
    MetricTriggerOperator,
    PrefectMetric,
    RunDeployment,
)


async def create_no_pending_automation(
    automation_name: str,
    target_flow_name: str,              # 被监控的 Flow 名称
    deployment_to_run: str,             # 要触发的部署，格式 "flow_name/deployment_name"
    window_seconds: int = 60,           # 条件需要维持成立的窗口（例如 60 秒）
    enabled: bool = True,
    description: str | None = None,
):
    """
    创建一个 Automation：
    条件：target_flow_name 的 Scheduled/Late 的运行个数 == 0（持续 window_seconds 秒）
    动作：RunDeployment(deployment_to_run)
    """
    # 构造 MetricTrigger 的查询
    # 提示：
    # - 如果运行时报 ValidationError，可能是你的 Server 小版本对字段名略有不同：
    #   1) 将 'value' 改成 'threshold'
    #   2) 或将 'window_seconds' 改成 'window'
    #   3) PrefectMetric 的枚举名称可能不同，可打印 list(PrefectMetric) 查看可选项
    query = MetricTriggerQuery(
        metric=PrefectMetric.flow_run_count,      # 常见可用：flow_run_count
        operator=MetricTriggerOperator.EQ,        # 等于
        value=0,                                  # 或者改用 threshold=0（如报错）
        window_seconds=window_seconds,            # 或者改用 window=window_seconds（如报错）
        # 过滤条件：目标 Flow 且 FlowRun 状态类型在 [SCHEDULED, LATE]
        filters={
            "flow": {
                "name": {"any_": [target_flow_name]}
            },
            "flow_run": {
                "state": {
                    "type": {"any_": ["SCHEDULED", "LATE"]}
                }
            },
        },
    )

    trigger = MetricTrigger(query=query)

    action = RunDeployment(
        # 一般支持用名称指定；如你更偏好 ID，可改用 deployment_id（取决于版本）
        deployment_name=deployment_to_run
    )

    automation = AutomationCore(
        name=automation_name,
        enabled=enabled,
        description=description
        or f"当 {target_flow_name} 不再有 Scheduled/Late 时触发 {deployment_to_run}",
        trigger=trigger,
        actions=[action],
    )

    async with get_client() as client:
        created = await client.create_automation(automation=automation)
        print(f"Automation 已创建：{created.name} (id={getattr(created, 'id', 'unknown')})")
        return created


if __name__ == "__main__":
    asyncio.run(
        create_no_pending_automation(
            automation_name="no-pending-then-run-downstream",
            target_flow_name="upstream-flow",                           # 替换成你的 Flow 名称
            deployment_to_run="downstream-flow/downstream-deployment",  # 替换成你的部署
            window_seconds=60,
        )
    )
使用与校验建议： - 运行脚本后，可用 
prefect automations ls
查看是否创建成功（CLI 目前没有 create，但能列出/查看/暂停/恢复）。 - 如果遇到 pydantic 的 ValidationError： - 把 
value=0
改为 
threshold=0
- 把 
window_seconds=...
改为 
window=...
- 如报 PrefectMetric 无该枚举，可临时打印 
print(list(PrefectMetric))
看你的环境有哪些可用枚举（通常 
flow_run_count
是可用的）。 - 也可把 
filters
中状态类型的值尝试小写（如 `["scheduled", "late"]`），但在 3.4.x 中大写更常见。 如果你愿意，把你这台 Server 的 API URL 发我（或直接打开 <你的API>/docs 搜索 “AutomationCreate” 与 “MetricTriggerQuery”），我可以根据你实例的 OpenAPI 精确对齐字段名，给你一份完全版零改动即可创建的脚本。 文档参考： - Automations 概览 - Events 概念
t
@Nate why is Marvin faster at giving out answers when it switch to a different language lmao
n
hm i didn’t read the thread but i assume it already did the tool calling when you asked in english so it probably didn’t have to do any more to translate the answer
t
ahhh I see I was inspired by the Portuguese thread
n
haha gpt-5 is mr worldwide indeed
🚀 1
t
@Marvin help it broke: 
refectHTTPStatusError: Client error '422 Unprocessable Entity' for url '<http://10.1.37.163:4200/api/automations/>'
Response: {'exception_message': 'Invalid request received.', 'exception_detail': [{'type': 'literal_error', 'loc': ['body', 'trigger', 'EventTrigger', 'type'], 'msg': "Input should be 'event'", 'input': 'metric', 'ctx': {'expected': "'event'"}}, {'type': 'missing', 'loc': ['body', 'trigger', 'EventTrigger', 'posture'], 'msg': 'Field required', 'input': {'type': 'metric', 'firing_for_seconds': 600, 'queries': [{'metric': 'prefect.flow_run.count', 'operator': '==', 'value': 0, 'where': {'flow.name': {'any_': ['inference']}, 'deployment.name': {'any_': ['inference']}, 'state.type': {'any_': ['SCHEDULED', 'LATE']}}}]}}, {'type': 'literal_error', 'loc': ['body', 'trigger', 'function-after[validate_require(), function-after[validate_require(), CompoundTrigger]]', 'type'], 'msg': "Input should be 'compound'", 'input': 'metric', 'ctx': {'expected': "'compound'"}}, {'type': 'missing', 'loc': ['body', 'trigger', 'function-after[validate_require(), function-after[validate_require(), CompoundTrigger]]', 'triggers'], 'msg': 'Field required', 'input': {'type': 'metric', 'firing_for_seconds': 600, 'queries': [{'metric': 'prefect.flow_run.count', 'operator': '==', 'value': 0, 'where': {'flow.name': {'any_': ['inference']}, 'deployment.name': {'any_': ['inference']}, 'state.type': {'any_': ['SCHEDULED', 'LATE']}}}]}}, {'type': 'missing', 'loc': ['body', 'trigger', 'function-after[validate_require(), function-after[validate_require(), CompoundTrigger]]', 'within'], 'msg': 'Field required', 'input': {'type': 'metric', 'firing_for_seconds': 600, 'queries': [{'metric': 'prefect.flow_run.count', 'operator': '==', 'value': 0, 'where': {'flow.name': {'any_': ['inference']}, 'deployment.name': {'any_': ['inference']}, 'state.type': {'any_': ['SCHEDULED', 'LATE']}}}]}}, {'type': 'missing', 'loc': ['body', 'trigger', 'function-after[validate_require(), function-after[validate_require(), CompoundTrigger]]', 'require'], 'msg': 'Field required', 'input': {'type': 'metric', 'firing_for_seconds': 600, 'queries': [{'metric': 'prefect.flow_run.count', 'operator': '==', 'value': 0, 'where': {'flow.name': {'any_': ['inference']}, 'deployment.name': {'any_': ['inference']}, 'state.type': {'any_': ['SCHEDULED', 'LATE']}}}]}}, {'type': 'literal_error', 'loc': ['body', 'trigger', 'SequenceTrigger', 'type'], 'msg': "Input should be 'sequence'", 'input': 'metric', 'ctx': {'expected': "'sequence'"}}, {'type': 'missing', 'loc': ['body', 'trigger', 'SequenceTrigger', 'triggers'], 'msg': 'Field required', 'input': {'type': 'metric', 'firing_for_seconds': 600, 'queries': [{'metric': 'prefect.flow_run.count', 'operator': '==', 'value': 0, 'where': {'flow.name': {'any_': ['inference']}, 'deployment.name': {'any_': ['inference']}, 'state.type': {'any_': ['SCHEDULED', 'LATE']}}}]}}, {'type': 'missing', 'loc': ['body', 'trigger', 'SequenceTrigger', 'within'], 'msg': 'Field required', 'input': {'type': 'metric', 'firing_for_seconds': 600, 'queries': [{'metric': 'prefect.flow_run.count', 'operator': '==', 'value': 0, 'where': {'flow.name': {'any_': ['inference']}, 'deployment.name': {'any_': ['inference']}, 'state.type': {'any_': ['SCHEDULED', 'LATE']}}}]}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'DoNothing', 'type'], 'msg': "Input should be 'do-nothing'", 'input': 'run-deployment', 'ctx': {'expected': "'do-nothing'"}}, {'type': 'value_error', 'loc': ['body', 'actions', 0, 'function-after[selected_deployment_requires_id(), RunDeployment]'], 'msg': 'Value error, deployment_id is required', 'input': {'type': 'run-deployment', 'deployment': 'add-inference-job-to-queue/add-inference-job-to-queue', 'parameters': {'update_db': True, 'model_path': '/root/vast/than/als2h_basemodel/0922CVATNN/models/Leopard', 'output_dir': '/root/vast/than/preds_v4', 'kwargs': {'batch_size': 8, 'queue_maxsize': 16, 'tracking': True, 'candidates_method': 'local_queues', 'max_tracks': 3}, 'max_jobs': 30}, 'tags': ['auto-triggered']}, 'ctx': {'error': {}}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'function-after[selected_deployment_requires_id(), PauseDeployment]', 'type'], 'msg': "Input should be 'pause-deployment'", 'input': 'run-deployment', 'ctx': {'expected': "'pause-deployment'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'function-after[selected_deployment_requires_id(), ResumeDeployment]', 'type'], 'msg': "Input should be 'resume-deployment'", 'input': 'run-deployment', 'ctx': {'expected': "'resume-deployment'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'CancelFlowRun', 'type'], 'msg': "Input should be 'cancel-flow-run'", 'input': 'run-deployment', 'ctx': {'expected': "'cancel-flow-run'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'ChangeFlowRunState', 'type'], 'msg': "Input should be 'change-flow-run-state'", 'input': 'run-deployment', 'ctx': {'expected': "'change-flow-run-state'"}}, {'type': 'missing', 'loc': ['body', 'actions', 0, 'ChangeFlowRunState', 'state'], 'msg': 'Field required', 'input': {'type': 'run-deployment', 'deployment': 'add-inference-job-to-queue/add-inference-job-to-queue', 'parameters': {'update_db': True, 'model_path': '/root/vast/than/als2h_basemodel/0922CVATNN/models/Leopard', 'output_dir': '/root/vast/than/preds_v4', 'kwargs': {'batch_size': 8, 'queue_maxsize': 16, 'tracking': True, 'candidates_method': 'local_queues', 'max_tracks': 3}, 'max_jobs': 30}, 'tags': ['auto-triggered']}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'function-after[selected_work_queue_requires_id(), PauseWorkQueue]', 'type'], 'msg': "Input should be 'pause-work-queue'", 'input': 'run-deployment', 'ctx': {'expected': "'pause-work-queue'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'function-after[selected_work_queue_requires_id(), ResumeWorkQueue]', 'type'], 'msg': "Input should be 'resume-work-queue'", 'input': 'run-deployment', 'ctx': {'expected': "'resume-work-queue'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'SendNotification', 'type'], 'msg': "Input should be 'send-notification'", 'input': 'run-deployment', 'ctx': {'expected': "'send-notification'"}}, {'type': 'missing', 'loc': ['body', 'actions', 0, 'SendNotification', 'block_document_id'], 'msg': 'Field required', 'input': {'type': 'run-deployment', 'deployment': 'add-inference-job-to-queue/add-inference-job-to-queue', 'parameters': {'update_db': True, 'model_path': '/root/vast/than/als2h_basemodel/0922CVATNN/models/Leopard', 'output_dir': '/root/vast/than/preds_v4', 'kwargs': {'batch_size': 8, 'queue_maxsize': 16, 'tracking': True, 'candidates_method': 'local_queues', 'max_tracks': 3}, 'max_jobs': 30}, 'tags': ['auto-triggered']}}, {'type': 'missing', 'loc': ['body', 'actions', 0, 'SendNotification', 'body'], 'msg': 'Field required', 'input': {'type': 'run-deployment', 'deployment': 'add-inference-job-to-queue/add-inference-job-to-queue', 'parameters': {'update_db': True, 'model_path': '/root/vast/than/als2h_basemodel/0922CVATNN/models/Leopard', 'output_dir': '/root/vast/than/preds_v4', 'kwargs': {'batch_size': 8, 'queue_maxsize': 16, 'tracking': True, 'candidates_method': 'local_queues', 'max_tracks': 3}, 'max_jobs': 30}, 'tags': ['auto-triggered']}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'CallWebhook', 'type'], 'msg': "Input should be 'call-webhook'", 'input': 'run-deployment', 'ctx': {'expected': "'call-webhook'"}}, {'type': 'missing', 'loc': ['body', 'actions', 0, 'CallWebhook', 'block_document_id'], 'msg': 'Field required', 'input': {'type': 'run-deployment', 'deployment': 'add-inference-job-to-queue/add-inference-job-to-queue', 'parameters': {'update_db': True, 'model_path': '/root/vast/than/als2h_basemodel/0922CVATNN/models/Leopard', 'output_dir': '/root/vast/than/preds_v4', 'kwargs': {'batch_size': 8, 'queue_maxsize': 16, 'tracking': True, 'candidates_method': 'local_queues', 'max_tracks': 3}, 'max_jobs': 30}, 'tags': ['auto-triggered']}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'function-after[selected_automation_requires_id(), PauseAutomation]', 'type'], 'msg': "Input should be 'pause-automation'", 'input': 'run-deployment', 'ctx': {'expected': "'pause-automation'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'function-after[selected_automation_requires_id(), ResumeAutomation]', 'type'], 'msg': "Input should be 'resume-automation'", 'input': 'run-deployment', 'ctx': {'expected': "'resume-automation'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'SuspendFlowRun', 'type'], 'msg': "Input should be 'suspend-flow-run'", 'input': 'run-deployment', 'ctx': {'expected': "'suspend-flow-run'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'ResumeFlowRun', 'type'], 'msg': "Input should be 'resume-flow-run'", 'input': 'run-deployment', 'ctx': {'expected': "'resume-flow-run'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'function-after[selected_work_pool_requires_id(), PauseWorkPool]', 'type'], 'msg': "Input should be 'pause-work-pool'", 'input': 'run-deployment', 'ctx': {'expected': "'pause-work-pool'"}}, {'type': 'literal_error', 'loc': ['body', 'actions', 0, 'function-after[selected_work_pool_requires_id(), ResumeWorkPool]', 'type'], 'msg': "Input should be 'resume-work-pool'", 'input': 'run-deployment', 'ctx': {'expected': "'resume-work-pool'"}}, {'type': 'extra_forbidden', 'loc': ['body', 'posture'], 'msg': 'Extra inputs are not permitted', 'input': 'Reactive'}], 'request_body': {'name': 'Run add-inference when inference has no pending runs for 10m', 'description': 'If inference/inference has no Scheduled or Late runs for 10 minutes, run add-inference-job-to-queue.', 'enabled': True, 'posture': 'Reactive', 'trigger': {'type': 'metric', 'firing_for_seconds': 600, 'queries': [{'metric': 'prefect.flow_run.count', 'operator': '==', 'value': 0, 'where': {'flow.name': {'any_': ['inference']}, 'deployment.name': {'any_': ['inference']}, 'state.type': {'any_': ['SCHEDULED', 'LATE']}}}]}, 'actions': [{'type': 'run-deployment', 'deployment': 'add-inference-job-to-queue/add-inference-job-to-queue', 'parameters': {'update_db': True, 'model_path': '/root/vast/than/als2h_basemodel/0922CVATNN/models/Leopard', 'output_dir': '/root/vast/than/preds_v4', 'kwargs': {'batch_size': 8, 'queue_maxsize': 16, 'tracking': True, 'candidates_method': 'local_queues', 'max_tracks': 3}, 'max_jobs': 30}, 'tags': ['auto-triggered']}]}}
m
Your message was too long, here's your message at the allowed limit: help it broke 
refectHTTPStatusError Client error '422 Unprocessable Entity' for url '<http //10 1 37 163 4200/api/automations/>'
`Response {'exception_message' 'Invalid request received ' 'exception_detail' [{'type' 'literal_error' 'loc' ['body' 'trigger' 'EventTrigger' 'type'] 'msg' Input should be 'event' 'input' 'metric' 'ctx' {'expected' 'event' }} {'type' 'missing' 'loc' ['body' 'trigger' 'EventTrigger' 'posture'] 'msg' 'Field required' 'input' {'type' 'metric' 'firing_for_seconds' 600 'queries' [{'metric' 'prefect flow_run count' 'operator' '==' 'value' 0 'where' {'flow name' {'any_' ['inference']} 'deployment name' {'any_' ['inference']} 'state type' {'any_' ['SCHEDULED' 'LATE']}}}]}} {'type' 'literal_error' 'loc' ['body' 'trigger' 'function-after[validate_require() function-after[validate_require() CompoundTrigger]]' 'type'] 'msg' Input should be 'compound' 'input' 'metric' 'ctx' {'expected' 'compound' }} {'type' 'missing' 'loc' ['body' 'trigger' 'function-after[validate_require() function-after[validate_require() CompoundTrigger]]' 'triggers'] 'msg' 'Field required' 'input' {'type' 'metric' 'firing_for_seconds' 600 'queries' [{'metric' 'prefect flow_run count' 'operator' '==' 'value' 0 'where' {'flow name' {'any_' ['inference']} 'deployment name' {'any_' ['inference']} 'state type' {'any_' ['SCHEDULED' 'LATE']}}}]}} {'type' 'missing' 'loc' ['body' 'trigger' 'function-after[validate_require() function-after[validate_require() CompoundTrigger]]' 'within'] 'msg' 'Field required' 'input' {'type' 'metric' 'firing_for_seconds' 600 'queries' [{'metric' 'prefect flow_run count' 'operator' '==' 'value' 0 'where' {'flow name' {'any_' ['inference']} 'deployment name' {'any_' ['inference']} 'state type' {'any_' ['SCHEDULED' 'LATE']}}}]}} {'type' 'missing' 'loc' ['body' 'trigger' 'function-after[validate_require() function-after[validate_require() CompoundTrigger]]' 'require'] 'msg' 'Field required' 'input' {'type' 'metric' 'firing_for_seconds' 600 'queries' [{'metric' 'prefect flow_run count' 'operator' '==' 'value' 0 'where' {'flow name' {'any_' ['inference']} 'deployment name' {'any_' ['inference']} 'state type' {'any_' ['SCHEDULED' 'LATE']}}}]}} {'type' 'literal_error' 'loc' ['body' 'trigger' 'SequenceTrigger' 'type'] 'msg' Input should be 'sequence' 'input' 'metric' 'ctx' {'expected' 'sequence' }} {'type' 'missing' 'loc' ['body' 'trigger' 'SequenceTrigger' 'triggers'] 'msg' 'Field required' 'input' {'type' 'metric' 'firing_for_seconds' 600 'queries' [{'metric' 'prefect flow_run count' 'operator' '==' 'value' 0 'where' {'flow name' {'any_' ['inference']} 'deployment name' {'any_' ['inference']} 'state type' {'any_' ['SCHEDULED' 'LATE']}}}]}} {'type' 'missing' 'loc' ['body' 'trigger' 'SequenceTrigger' 'within'] 'msg' 'Field required' 'input' {'type' 'metric' 'firing_for_seconds' 600 'queries' [{'metric' 'prefect flow_run count' 'operator' '==' 'value' 0 'where' {'flow name' {'any_' ['inference']} 'deployment name' {'any_' ['inference']} 'state type' {'any_' ['SCHEDULED' 'LATE']}}}]}} {'type' 'literal_error' 'loc' ['body' 'actions' 0 'DoNothing' 'type'] 'msg' Input should be 'do-nothing' 'input' 'run-deployment' 'ctx' {'expected' 'do-nothing' }} {'type' 'value_error' 'loc' ['body' 'actions' 0 'function-after[selected_deployment_requires_id() RunDeployment]'] 'msg' 'Value error deployment_id is required' 'input' {'type' 'run-deployment' 'deployment' 'add-inference-job-to-queue/add-inference-job-to-queue' 'parameters' {'update_db' True 'model_path' '/root/vast/than/als2h_basemodel/0922CVATNN/models/Leopard' 'output_dir' '/root/vast/than/preds_v4' 'kwargs' {'batch_size' 8 'queue_maxsize' 16 'tracking' True 'candidates_method' 'local_queues' 'max_tracks' 3} 'max_jobs' 30} 'tags' ['auto-triggered']} 'ctx' {'error' {}}} {'type' 'literal_error' 'loc' ['body' 'actions' 0 'function-after[selected_deployment_requires_id() PauseDeployment]' 'type'] 'msg' Input should be
'pause-deployment' 'input' 'run-deployment' 'ctx' {'expected' 'pause-deployment' }} {'type' 'literal_error' 'loc' ['body' 'actions' 0 'function-after[selected_deployment_requires_id() ResumeDeployment]' 'type'] 'msg' Input should be 'resume-deployment' 'input' 'run-deployment' 'ctx' {'expected' 'resume-deployment' }} {'type' 'literal_error' 'loc' ['body' 'actions' 0 'CancelFlowRun' 'type'] 'msg' Input should be 'cancel-flow-run' 'input' 'run-deployment' 'ctx' {'expected' 'cancel-flow-run' }} {'type' 'literal_error' 'loc' ['body' 'actions' 0 'ChangeFlowRunState' 'type'] 'msg' Input should be 'change-flow-run-state' 'input' 'run-deployment' 'ctx' {'expected' 'change-flow-run-state' }} {'type' 'missing' 'loc' ['body' 'actions' 0 'ChangeFlowRunState' 'state'] 'msg' 'Field required' 'input' {'type' 'run-deployment' 'deployment' 'add-inference-job-to-queue/add-inference-job-to-queue' 'parameters' {'update_db' True 'model_path' '/root/vast/than/als2h_basemodel/0922CVATNN/models/Leopard' 'output_dir' '/root/vast/than/preds_v4' 'kwargs' {'batch_size' 8 'queue_maxsize' 16 'tracking' True 'candidates_method' 'local_queues' 'max_tracks' 3} 'max_jobs' 30} 'tags' ['auto-triggered']}} {'type' 'literal_error' 'loc' ['body' 'actions' 0 'function-after[selected_work_queue_requires_id()
4 Views
Open in Slack
PreviousNext
Powered by