How to emit and use custom events

Learn how to define specific trigger conditions based on custom event grammar.

​

Motivating custom events

Imagine you are running an e-commerce platform and you want to trigger a deployment when a customer completes an order.

There might be a number of events that occur during an order on your platform, for example:

• order.created
• order.item.added
• order.payment-method.confirmed
• order.shipping-method.added
• order.complete

Event grammar

These event names are only an example. Choose any grammar that fits your use case, as long as names don’t start with prefect. or prefect-cloud., which are reserved for events that Prefect emits; see namespace conventions.

In this case, we want to trigger a deployment when a user completes an order, so our trigger should:

• expect an order.complete event
• after an order.created event
• evaluate these conditions for_each user id

Finally, it should pass the user_id as a parameter to the deployment.

​

Define the trigger

Here’s how this looks in code:

post_order_deployment.py

from prefect import flow
from prefect.events.schemas.deployment_triggers import DeploymentEventTrigger

order_complete = DeploymentEventTrigger(
    expect={"order.complete"},
    after={"order.created"},
    for_each={"prefect.resource.id"},
    parameters={"user_id": "{{ event.resource.id }}"},
)


@flow(log_prints=True)
def post_order_complete(user_id: str):
    print(f"User {user_id} has completed an order -- doing stuff now")


if __name__ == "__main__":
    post_order_complete.serve(triggers=[order_complete])

Specify multiple events or resources

The expect and after fields accept a set of event names, so you can specify multiple events for each condition.

Similarly, the for_each field accepts a set of resource ids.

​

Simulate events

To simulate users causing order status events, run the following in a Python shell or script:

simulate_events.py

from prefect.events import emit_event

user_id_1, user_id_2 = "123", "456"

order_created_1 = emit_event(
    event="order.created",
    resource={"prefect.resource.id": user_id_1},
)
emit_event(
    event="order.created",
    resource={"prefect.resource.id": user_id_2},  # other user
)
emit_event(
    event="order.complete",
    resource={"prefect.resource.id": user_id_1},
    follows=order_created_1,
)

In the above example:

• user_id_1 creates and then completes an order, triggering a run of our deployment.
• user_id_2 creates an order, but no completed event is emitted so no deployment is triggered.

Declare follows for events emitted close together

Because the trigger uses order.created in after and order.complete in expect, it only counts an order.complete event that arrives after an order.created event. Events emitted within a second or two of each other may arrive at the system in either order unless the later event declares follows—which is why the order.complete emission above passes follows=order_created_1. In a real system where minutes pass between creating and completing an order, follows isn’t strictly needed, though it still documents the relationship between the events (note that emit_event only sets it when the two events occur within five minutes of each other).