https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I am using a webhook, but it has to b...
# ask-marvin
m
@Marvin I am using a webhook, but it has to be a static one, since i can't json parse the body of the message being send to the webhook. How can i create an automation, that parses the body-string of the event and run the automation only, if a specific word is in the body string?
m
In Prefect, the body of a webhook goes through some light preprocessing to make it more useful in templates. If the Content-Type of the request is application/json, the body will be parsed as a JSON object and made available to the webhook templates. If the Content-Type is application/x-www-form-urlencoded, the body is parsed into a flat dict-like object of key-value pairs. To create an automation that parses the body-string of the event and run the automation only if a specific word is in the body string, you can use a conditional statement in your automation script to check if the specific word is in the parsed body-string. If the word is found, then the automation is run. Here's a simple example:
Copy code
# Assume `body_string` is the parsed body-string of your webhook event
if 'specific_word' in body_string:
    # Run your automation here
Please replace 'specific_word' with the actual word you want to check in the body-string. Also, make sure to replace the comment inside the if statement with the actual code of your automation. For more information, you can refer to the Prefect's webhooks guide.
m
@Marvin I would like to use the prefect automation functionality and need to fill out the following, can you help me with this, is it possible to trigger the automation only if a specific word is in the event body string?: { "match": { "prefect.resource.id": "prefect.flow-run.*" }, "match_related": { "prefect.resource.id": "prefect.deployment.70cb25fe-e33d-4f96-b1bc-74aa4e50b761", "prefect.resource.role": "deployment" }, "for_each": [ "prefect.resource.id" ], "after": [], "expect": [ "prefect.flow-run.Failed" ], "posture": "Reactive", "threshold": 2, "within": 10 }
m
In Prefect, the automation functionality is designed to react to specific events. The automation configuration you provided is set to trigger when a flow run fails (
"expect": ["prefect.flow-run.Failed"]
) for a specific deployment (
"prefect.resource.id": "prefect.deployment.70cb25fe-e33d-4f96-b1bc-74aa4e50b761"
). However, the current automation functionality does not support triggering based on the content of the event body string. The automation is designed to react to specific events and their associated metadata, not the content of the event body. If you need to trigger an action based on the content of the event body, you might need to handle this in the flow itself, or in a separate service that listens to the webhook events and triggers the appropriate action based on the content of the event body. For more information, you can refer to the Prefect's automation guide.
n
Copy code
» python examples/flows/on_event_body.py
╭──────────────────────────────────────────────────────────────────────────────────
│ Your deployments are being served and polling for scheduled runs!                                                                                                              │
│         Deployments                                                                                                                                                            │
│ ┌──────────────────────────┐                                                                                                                                                   │
│ │ downstream/on_event_body │                                                                                                                                                   │
│ └──────────────────────────┘                                                                                                                                                   │
│ To trigger any of these deployments, use the following command:                                                                                                                │
│                                                                                                                                                                                │
│         $ prefect deployment run [DEPLOYMENT_NAME]                                                                                                                             │
│                                                                                                                                                                                │
│ You can also trigger your deployments via the Prefect UI:                                                                                                                      │
│ <https://app.prefect.cloud/account/xxx/workspace/xxx/deployments>                                              

00:40:37.604 | INFO    | prefect.flow_runs.runner - Runner 'runner-690f5a56-60b3-4eaa-9fb8-3b48ab49cf41' submitting flow run 'b58954e4-fea4-408b-a48e-5780014794bb'
00:40:37.749 | INFO    | prefect.flow_runs.runner - Opening process...
00:40:37.757 | INFO    | prefect.flow_runs.runner - Completed submission of flow run 'b58954e4-fea4-408b-a48e-5780014794bb'
00:40:39.457 | INFO    | Flow run 'camouflaged-pigeon' - Downloading flow code from storage at '.'
00:40:39.971 | INFO    | Flow run 'camouflaged-pigeon' - i was triggered!
00:40:40.097 | INFO    | Flow run 'camouflaged-pigeon' - Finished in state Completed()
00:40:40.510 | INFO    | prefect.flow_runs.runner - Process for flow run 'camouflaged-pigeon' exited cleanly.
on_event.py
i would just add a label to 
match
on within the 
resource
instead of trying to do conditional stuff on the event 
payload
since you can add / match on whatever labels you want in the automation, also my use of the sdk for the trigger instead of yaml and my use of 
serve
for the deployment is arbitrary, could easily do this in 
prefect.yaml
as well if you're wanting to react to a specific part of 
prefect.flow-run.Failed
- you may want to manually 
emit_event
from an 
on_failure
hook on the offending flow
m
@Nate Hi Nate, thank you very much for the fast response, I didn't even expect to "bother" some human with my question here in the ask-marvin channel :) Essentially my problem is, that I get 3 events from the same source on my prefect webhook, but I need only the last one, which has a specific keyword in the event msg. Since my event msg is a xml, which has colons, etc in the tags like this example: <env:Envelope xmlns:env=\"http://www.w3.org/2003/05/soap-envelope\"> I wasn't able to parse it to find and assign the keyword to a resource parameter in the event json object, which I think I could use to match my prefect automation on. Do you know if this should be possible (parsing the xml)? If yes, I would work further on this approach ... Thank you very much for showing me your approach, which I think is very interesting. But I did not get, how I can parse my event data here, do I have access to the event body parameter? Do I get it correct, that you simulate an incoming event in your example by emitting an event with emit_event(..). And you are deploying the downstream flow to be triggered by the deployment trigger, which is looking for your emitted event. But in your example the random_data does not come from an incoming event, so I can't distinguish between the 3 events ether ... I think I am missing something here ...
@Marvin I am trying to create a prefect webhook which parses an incoming event . The event is a xml message and has colons, etc in its tags like this example: <env:Envelope xmlns:env=\"http://www.w3.org/2003/05/soap-envelope\">, is it possible with the prefect webhook to parse this kind of xml into a json event object?
m
Prefect's webhook functionality is designed to handle JSON payloads and URL-encoded forms. It uses Jinja2 templates to parse the body of the incoming HTTP request and generate events based on the parsed data. However, it does not natively support XML payloads. If your incoming event is an XML message, you would need to convert this XML to JSON or a URL-encoded form before sending it to the Prefect webhook. There are many online tools and libraries available that can convert XML to JSON. You could use one of these tools to convert your XML message to JSON before sending it to the Prefect webhook. Once the XML is converted to JSON, you can use Jinja2 templates in your Prefect webhook to parse the JSON and generate events based on the parsed data. For more information, you can refer to the Prefect's webhooks guide.
m
@Marvin how do I assign the full event payload body string to a resource parameter in a prefect webhook template?
m
In Prefect, you can assign the full event payload body string to a resource parameter in a webhook template using the 
body
variable. The 
body
variable contains the entire body of the incoming HTTP request. Here is an example of how you can do this:
Copy code
json
{
  "event": "my.event",
  "resource": {
    "prefect.resource.id": "my.resource",
    "prefect.resource.name": "My Resource",
    "my_parameter": "{{ body }}"
  }
}
In this example, the 
my_parameter
resource parameter is assigned the full event payload body string. Please note that the 
body
variable contains the body of the incoming HTTP request as a string. If the body was sent as JSON, you can convert it to a dictionary using the 
tojson
filter, like this: 
{{ body|tojson }}
. For more information, you can refer to the Prefect's webhooks guide.
m
@Marvin ok, but I get the error:
Copy code
"unexpected character: line 6 column 50 (char 201)"
it seems that the webhook still tries to parse my xml msg ... how can i just access my message as string from out an event? should i use a static webhook? how can I pass the event payload body into my flow with an prefect automation?
m
It seems like you're running into issues because Prefect is trying to parse the XML as JSON. Prefect's webhooks are designed to handle JSON payloads and URL-encoded forms, and they use Jinja2 templates to parse the body of the incoming HTTP request. If you're receiving XML payloads, you might need to handle the parsing within your flow. You can pass the entire payload as a string to your flow using Prefect's automation functionality. Here's a general idea of how you can do this: 1. Create a static webhook in Prefect. This will allow you to receive the XML payload without Prefect trying to parse it. 2. In your automation configuration, set up an action to start a flow run and pass the entire payload as a parameter to your flow. Here's an example of how you can do this:
Copy code
json
{
  "match": {
    "prefect.resource.id": "prefect.flow-run.*"
  },
  "match_related": {
    "prefect.resource.id": "prefect.deployment.70cb25fe-e33d-4f96-b1bc-74aa4e50b761",
    "prefect.resource.role": "deployment"
  },
  "for_each": [
    "prefect.resource.id"
  ],
  "after": [],
  "expect": [
    "prefect.flow-run.Failed"
  ],
  "posture": "Reactive",
  "threshold": 2,
  "within": 10,
  "actions": [
    {
      "type": "prefect.flow-run.create",
      "config": {
        "flow_id": "your-flow-id",
        "parameters": {
          "payload": "{{ body }}"
        }
      }
    }
  ]
}
In this example, the 
payload
parameter of your flow will receive the entire XML payload as a string. 3. In your flow, you can then parse the XML payload as needed. There are many Python libraries available that can parse XML, such as 
xml.etree.ElementTree
and 
lxml
. Please replace 
"your-flow-id"
with the actual ID of your flow. For more information, you can refer to the Prefect's automation guide and webhooks guide.
m
@Marvin are you sure I just have to pass in {{ body }} ? Shouldn't it be {{ event.payload.body }} ?
m
You're correct, my apologies for the confusion. In the context of a Prefect webhook, the payload of the event can be accessed using 
{{ event.payload.body }}
. Here's the corrected automation configuration:
Copy code
json
{
  "match": {
    "prefect.resource.id": "prefect.flow-run.*"
  },
  "match_related": {
    "prefect.resource.id": "prefect.deployment.70cb25fe-e33d-4f96-b1bc-74aa4e50b761",
    "prefect.resource.role": "deployment"
  },
  "for_each": [
    "prefect.resource.id"
  ],
  "after": [],
  "expect": [
    "prefect.flow-run.Failed"
  ],
  "posture": "Reactive",
  "threshold": 2,
  "within": 10,
  "actions": [
    {
      "type": "prefect.flow-run.create",
      "config": {
        "flow_id": "your-flow-id",
        "parameters": {
          "payload": "{{ event.payload.body }}"
        }
      }
    }
  ]
}
In this example, the 
payload
parameter of your flow will receive the entire XML payload as a string. Please replace 
"your-flow-id"
with the actual ID of your flow. For more information, you can refer to the Prefect's automation guide and webhooks guide.
m
It works 👍
5 Views
Open in Slack
PreviousNext
Powered by