Kahu 22
09/17/2025, 7:52 AM{
"id": "019956a3-c8dc-7e4f-a713-102ef7258136",
"event": "axos.gold.table.updated",
"occurred": "2025-09-17T07:46:38.427Z",
"payload": {
"table_name": "gold.invoice_monthly_summary",
and this calls a trigger to update something else, is it possible to pass a value from the event to the automation as a param for the deployment which takes table_name as a payload or is this not possibleMarvin
09/17/2025, 7:52 AMMarvin
09/17/2025, 7:56 AM{{ event.payload.<key> }}
.
- For your example, use {{ event.payload.table_name }}
.
If you’re configuring the Automation in the UI
- Add an action: Run deployment
- In Parameters, add a key table_name
- Click the template toggle (</>) and set the value to:
{{ event.payload.table_name }}
If you’re defining the Automation as JSON (or via API), templated fields use the special JSON form:
"parameters": {
"table_name": {
"__prefect_kind": "jinja",
"template": "{{ event.payload.table_name }}"
}
}
Notes
- Make sure your deployment’s flow has a table_name
parameter in its signature.
- You can add simple logic/defaults in the template, e.g.:
{{ event.payload.table_name or 'unknown_table' }}
Docs
- Using templates in Automations
- Available Jinja filters and context
If you share how you’re creating the Automation (UI vs code) I can provide the exact steps or a ready-to-paste snippet.Kahu 22
09/18/2025, 3:26 AM{
"id": "019956b4-18a6-7c81-bcbf-47a93ad40c0a",
"event": "axos.gold.table.updated",
"occurred": "2025-09-17T08:04:27.429Z",
"payload": {
"table_name": "gold.invoice_monthly_summary",
"timestamp": "2025-09-17T20:04:27.428100"
},
"received": "2025-09-17T08:04:27.466Z",
"related": [
{
"prefect.resource.id": "prefect.flow-run.37b3953c-33f8-49a3-8905-6b1f7f19ce33",
"prefect.resource.role": "flow-run",
"prefect.resource.name": "clever-whippet"
},
{
"prefect.resource.id": "prefect.task-run.019956b3-9734-7fd7-9dad-da2a5a996c9d",
"prefect.resource.role": "task-run",
"prefect.resource.name": "InvoiceMonthlySummaryGoldWorker-644"
},
{
"prefect.resource.id": "prefect.flow.8d5bdcf8-db0d-4a9e-ad44-2816ade6496b",
"prefect.resource.role": "flow",
"prefect.resource.name": "silver-to-gold-pipeline"
}
],
"resource": {
"prefect.resource.id": "gold.invoice_monthly_summary",
"table": "gold.invoice_monthly_summary",
"worker": "invoice_monthly_summary"
}
}
i need this to be caught through a trigger and call my dashboard-widget-update deployment. this deployment takes a table_name as a key, as shown by the python code that it is connected to:
@flow(name="update-dashboard-widget")
def update_dashboard_widget(table_name):
i would like the steps to create this in the ui, so i can pass the table name from the event to the deployment.Marvin
09/18/2025, 3:26 AMMarvin
09/18/2025, 3:27 AMtable_name
from the event payload to your deployment.
Here are the UI steps in Prefect 3.x:
1) Open Automations
- In the Prefect UI, go to Automations.
- Click “New automation”.
2) Name and Trigger
- Give it a name like “Update dashboard widget on gold table update”.
- Trigger: choose “Event occurs”.
- Condition: set Event name equals axos.gold.table.updated
.
3) Action: Run deployment
- Add action: “Run deployment”.
- Select your deployment: “dashboard-widget-update” (or the exact name shown in your Deployments list).
- Parameters:
- Add a parameter key: table_name
- Click the template toggle (</>) for the value field
- Enter:
{{ event.payload.table_name }}
- Save the action.
4) Save and test
- Save the automation.
- When the event you showed is emitted, the automation will run the deployment and pass the value gold.invoice_monthly_summary
into your flow’s table_name
argument (matching your flow signature def update_dashboard_widget(table_name):
).
Notes and options:
- Your event also includes resource.table
. If you prefer that source, you can use:
{{ event.resource.table }}
- Make sure your deployment’s flow parameter is named exactly table_name
. If it differs, update either the flow signature or the parameter key in the action.
Docs
- Using templates in Automations
- Available Jinja filters/context
If you share a screenshot of your deployment’s parameters panel, I can double-check the exact field names and confirm the template entry for you.Kahu 22
09/18/2025, 3:36 AMMarvin
09/18/2025, 3:36 AMMarvin
09/18/2025, 3:36 AMtable_name
- Click the template toggle (</>) for the value
- Paste:
{{ event.payload.table_name }}
- Save
If you prefer “Use JSON input”, the equivalent is:
{
"table_name": {
"__prefect_kind": "jinja",
"template": "{{ event.payload.table_name }}"
}
}
Docs:
- Using templates in Automations
- Available Jinja filters/contextKahu 22
09/18/2025, 8:24 PM{
"table_name": {
"__prefect_kind": "jinja",
"template": "{{ event.payload.table_name }}"
}
}
job variables:
{}
trigger:
{
"type": "event",
"match": {},
"match_related": {},
"after": [],
"expect": [
"axos.gold.table.updated"
],
"for_each": [],
"posture": "Reactive",
"threshold": 1,
"within": 10
}
action: run deploymnet
Deployment:
update-dashboard-widget > dashboard-widget-update
name:
update dashboard trigger
----
i want to make this reproducible through a run once script, how can i do this in python so this automation is created (currently have my work pool and deployments automatically created through my script)
currently i execute commands like so so what prams would i pass:
def run_deploy_command(command, description):
"""Run a deployment command"""
try:
result = subprocess.run(command, shell=True, capture_output=True, text=True, cwd=os.getcwd())
Marvin
09/18/2025, 8:24 PMMarvin
09/18/2025, 8:33 PMprefect automation create ...
command to your subprocess runner. Instead, create the automation via Python using the Prefect client.
Below is a minimal async script that:
- looks up your deployment by name (update-dashboard-widget/dashboard-widget-update
)
- creates an event-triggered automation with your Jinja parameter mapping for table_name
- sets the action to “Run deployment” with empty job variables
You can call this from your “run once” script.
import asyncio
from prefect.client.orchestration import get_client
from prefect.events.schemas.automations import AutomationCore, EventTrigger, Posture
from prefect.events.actions import RunDeployment
FLOW_NAME = "update-dashboard-widget"
DEPLOYMENT_NAME = "dashboard-widget-update"
AUTOMATION_NAME = "update dashboard trigger"
async def create_automation():
async with get_client() as client:
# Resolve deployment id by flow and deployment name
deployment = await client.read_deployment_by_name(
name=DEPLOYMENT_NAME,
flow_name=FLOW_NAME,
)
if deployment is None:
raise RuntimeError(f"Deployment not found: {FLOW_NAME}/{DEPLOYMENT_NAME}")
# Build the automation definition
automation = AutomationCore(
name=AUTOMATION_NAME,
enabled=True,
trigger=EventTrigger(
match={},
match_related={},
after=[],
expect=["axos.gold.table.updated"],
for_each=[],
posture=Posture.Reactive,
threshold=1,
within=10,
),
actions=[
RunDeployment(
deployment_id=deployment.id,
parameters={
"table_name": {
"__prefect_kind": "jinja",
"template": "{{ event.payload.table_name }}",
}
},
job_variables={},
)
],
)
created = await client.create_automation(automation)
print(f"Created automation: {created.name} ({created.id})")
if __name__ == "__main__":
asyncio.run(create_automation())
Notes:
- This assumes your script is already authenticated to Prefect Cloud (via PREFECT_API_KEY
and PREFECT_API_URL
) and the correct workspace is set.
- If you want the action to reference the deployment by name instead of id, we can adjust the payload, but using deployment_id
is the most reliable.
- The trigger body mirrors what you pasted: type: "event"
, posture: "Reactive"
, expect: ["axos.gold.table.updated"]
, etc.
If you really want to stick with your subprocess.run(...)
approach, you could wrap this Python in a standalone file and call python path/to/create_automation.py
from your script. There isn’t a supported prefect automation create
CLI in 3.x today, so Python is the recommended route.
Would you like me to add existence-check/update logic so the script is idempotent (create if missing, update if present)?