<@ULVA73B9P> I have three deployments: - *Flow 1: ...
# ask-marvin
b
@Marvin I have three deployments: • Flow 1: A (runs every day at 8:00) • Flow 2: B (runs every day at 15:00) • Flow 3: C I want to use Prefect automation to trigger deployment C only when both deployments A and B have successfully run and completed on the same day. How can I set this up?
m
thought for 251.3 seconds
Great question — you can do this with a single Prefect Automation using event triggers in Prefect 3.x. Goal: Run deployment C only after both A and B have completed successfully on the same day. Two good patterns (pick one): 1) Sequence trigger (recommended for your schedule) - Ensures order: A completes, then B completes - Add a time window so both completions must occur within the same day How to configure in the UI: - Go to Automations → New automation - Posture: Reactive - Trigger: Sequence - Step 1: Event = Flow run Completed - Filter by Deployment = A - Step 2: Event = Flow run Completed - Filter by Deployment = B - Window: within 1 day (or tighter, e.g., 18 hours, since A is 08:00 and B is 15:00) - Action: Run a deployment → select deployment C - Save This will fire only when B completes after A, within the window, which matches your “same day” requirement. 2) Compound (all-of) trigger - Does not enforce order, just requires both to complete within a time window How to configure in the UI: - Go to Automations → New automation - Posture: Reactive - Trigger: Compound - Child trigger 1: Event = Flow run Completed; Filter by Deployment = A - Child trigger 2: Event = Flow run Completed; Filter by Deployment = B - Require: All - Window: within 1 day (or tighter as above) - Action: Run a deployment → select deployment C - Save Notes and tips - “Completed” ensures only successful runs of A and B count. - The “within” window is a rolling duration, not tied to a specific calendar day. Given A=08:00 and B=15:00, a 1-day window is fine; you can tighten it if you want extra safety. - If you ever have multiple runs per day (manual reruns), the automation will trigger once per pair that satisfies the trigger logic. - If you need to pass parameters from A/B to C, that’s not automatic. You’d either: - Give C fixed/default params in its deployment, or - Orchestrate via a “controller” flow that runs A and B and then C, passing values explicitly. Docs to dig deeper - Automations and triggers overview: Creating automations and triggers - Event-driven flows with Prefect: Event-driven flows blog Want me to draft this automation for you in your workspace if you share the exact deployment names/IDs and any parameters C needs?
b
@Marvin i can not find any
Posture
in Prefect UI, can you give me an json file or yaml, python code for this set up?
m
thought for 523.3 seconds
Totally—here are ready-to-use examples to create this automation without the UI. In Prefect 3.x you can create automations via the REST API. Below are: - JSON payloads you can POST - curl and Python examples to create the automation - Both a Sequence (A then B) and a Compound (A and B in any order) version Notes - Use deployment IDs for reliability. You can get them with:
Copy code
prefect deployment inspect "flow-a/deployment-a"
  prefect deployment inspect "flow-b/deployment-b"
  prefect deployment inspect "flow-c/deployment-c"
- “Same day” is best approximated with a time window (
within
). Given A=08:00 and B=15:00, use
PT18H
or
PT24H
as appropriate. - Posture: You don’t need to set it; this is a reactive automation by default. Option 1: Sequence trigger (A completes, then B completes, within a window) JSON payload (replace placeholders):
Copy code
{
  "name": "Run C after A then B complete same day",
  "description": "Trigger deployment C only after A completes and then B completes within the time window.",
  "enabled": true,
  "trigger": {
    "type": "sequence",
    "within": "PT24H",
    "triggers": [
      {
        "type": "event",
        "match": {
          "event": "prefect.flow-run.Completed"
        },
        "match_related": [
          {
            "prefect.resource.id": "prefect.deployment.<DEPLOYMENT_ID_A>"
          }
        ]
      },
      {
        "type": "event",
        "match": {
          "event": "prefect.flow-run.Completed"
        },
        "match_related": [
          {
            "prefect.resource.id": "prefect.deployment.<DEPLOYMENT_ID_B>"
          }
        ]
      }
    ]
  },
  "actions": [
    {
      "type": "run-deployment",
      "deployment_id": "<DEPLOYMENT_ID_C>",
      "parameters": {}
    }
  ]
}
Option 2: Compound trigger (A and B complete in any order within a window)
Copy code
{
  "name": "Run C after A and B complete same day (any order)",
  "description": "Trigger deployment C after both A and B complete successfully within the time window.",
  "enabled": true,
  "trigger": {
    "type": "compound",
    "require": "all",
    "within": "PT24H",
    "triggers": [
      {
        "type": "event",
        "match": {
          "event": "prefect.flow-run.Completed"
        },
        "match_related": [
          {
            "prefect.resource.id": "prefect.deployment.<DEPLOYMENT_ID_A>"
          }
        ]
      },
      {
        "type": "event",
        "match": {
          "event": "prefect.flow-run.Completed"
        },
        "match_related": [
          {
            "prefect.resource.id": "prefect.deployment.<DEPLOYMENT_ID_B>"
          }
        ]
      }
    ]
  },
  "actions": [
    {
      "type": "run-deployment",
      "deployment_id": "<DEPLOYMENT_ID_C>",
      "parameters": {}
    }
  ]
}
How to create the automation (curl) - Set
PREFECT_API_URL
and
PREFECT_API_KEY
(Cloud) in your environment. - Then:
Copy code
curl -X POST "$PREFECT_API_URL/automations/" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d @automation.json
Where
automation.json
is one of the payloads above. How to create the automation (Python) ``` import os import httpx PREFECT_API_URL = os.environ.get("PREFECT_API_URL", "http://127.0.0.1:4200/api") PREFECT_API_KEY = os.environ.get("PREFECT_API_KEY") # Only needed for Cloud payload = { "name": "Run C after A then B complete same day", "enabled": True, "trigger": { "type": "sequence", "within": "PT24H", "triggers": [ { "type": "event", "match": {"event": "prefect.flow-run.Completed"}, "match_related": [{"prefect.resource.id": "prefect.deployment.<DEPLOYMENT_ID_A>"}] }, { "type": "event", "match": {"event": "prefect.flow-run.Completed"}, "match_related": [{"prefect.resource.id": "prefect.deployment.<DEPLOYMENT_ID_B>"}] } ] }, "actions": [
{"type": "run-deployment", "deployment_id": "<DEPLOYMENT_ID_C>", "parameters": {}} ] } headers = {"Content-Type": "application/json"} if PREFECT_API_KEY: headers["Authorization"] = f"Bearer {PREFECT_API_KEY}" resp = httpx.post(f"{PREFECT_API_URL}/automations/", json=payload, headers=headers, timeout=30) resp.raise_for_status() print("Created automation:", resp.json())
Copy code
How to get deployment IDs
- Replace with your names:
prefect deployment inspect "flow-a/deployment-a" prefect deployment inspect "flow-b/deployment-b" prefect deployment inspect "flow-c/deployment-c" ``` Copy the
id
from each and substitute into the JSON. A few practical tips - Tighten the window: If you strictly want “same day” in your local timezone, set a tighter
within
(e.g.,
PT18H
for 08:00 → 15:00) to prevent cross-midnight pairing. - Retries/reruns: Each pair that satisfies the trigger may cause a run of C; if you want just one, keep A and B on single daily schedules and avoid manual reruns, or consider adding a small controller flow. Docs - Creating automations and triggers: Creating automations and triggers - Event triggers overview: Event triggers concept If you share the actual deployment names/IDs for A, B, and C, I can fill in a final payload for you.
b
@Marvin thanks, in actions tag, what does it mean “Parameters defined for an automation action will be merged with the deployment’s default parameters when the action is executed.“?
m
thought for 37.5 seconds
Great question. It means: - When a RunDeployment action includes a
parameters
dict, Prefect will take those values and merge them over the deployment’s default parameters. - The merge is shallow (top-level keys only): - If you provide a key in the action’s
parameters
, it overrides the deployment’s default for that key. - If you omit a key, the deployment’s default for that key is used. - If you set a key to
null
(None), that null will override the default (i.e., the value becomes null). - Nested dicts are not deeply merged; a provided nested dict replaces the entire value for that key. Example - Deployment C defaults:
Copy code
{
  "date": "today",
  "region": "us-east-1",
  "options": {
    "full_refresh": false,
    "batch_size": 1000
  }
}
- Automation action parameters:
Copy code
{
  "region": "eu-west-1",
  "options": {
    "full_refresh": true
  }
}
- Effective parameters at runtime:
Copy code
{
    "date": "today",                 # from defaults (not overridden)
    "region": "eu-west-1",           # overridden by action
    "options": {"full_refresh": true}# replaced entirely (batch_size is NOT retained)
  }
Rule of thumb: - Provide only the keys you want to change. - If you need to keep nested defaults, include them explicitly in the action payload (since it’s a replace, not a deep-merge).
b
@Marvin Hi, in prefect how can i create trigger if a flow run from my deployment is Completed? Give me a json and yaml config. thanks
m
🔄 Researching your answer... this may take a while 🔧 Using tool:
research_prefect_topic
📊 Tools used: •
research_prefect_topic
(1x)