How to create event based workflow?

Community Support
1

Hi!
I just started with Temporal.io, and i feel lost.
Noob question: it is possible to create event based workflows?
The scenario is: start worflow with some actvities, and between this activities, i must wait for events (this events can come from webhook for example) to continue workflow, and it must be possible to spawn a lot of workflows like this.

I dont know if is similar to Background Checks example in Docs, but i know nothing about Go.

For example:
START → activity1 → wait event1 → activity2 → wait event2 → activity3 → wait event3 → activity4 → END

2

It is possible to create event workflows. For instance, you could create an asyncio.Event on self in __init__() and await wait() it during run() when you want to wait, and then set() it during a signal handler. Then you can send a signal to the workflow to trigger the event.

Does that make sense?

3

I tried this approach, but when i start the workflow, because of the wait(), my script going to run until the signal occurs. Suppose i use mensage broker like RabbitMQ to start this workflow in a consumer. I need to consume the message, start the workflow, and ack the message. I can’t lock the execution in the consumer to wait the event.

4

This is what you want and how workflows work. Workflows are built to wait for a long period of time.

For example (untested, just typed out here in chat):

@workflow.defn
class MyWorkflow:
    def __init__(self):
        self.events = [asyncio.Event(), asyncio.Event(), asyncio.Event()]

    @workflow.run
    async def run(self):
        workflow.logger.info("Executing activity 1")
        await workflow.execute_activity(my_activity_1)
        workflow.logger.info("Finished activity 1")
        await self.events[0].wait()
        
        workflow.logger.info("Executing activity 2")
        await workflow.execute_activity(my_activity_2)
        workflow.logger.info("Finished activity 2")
        await self.events[1].wait()
        
        workflow.logger.info("Executing activity 3")
        await workflow.execute_activity(my_activity_3)
        workflow.logger.info("Finished activity 3")
        await self.events[2].wait()
        
        workflow.logger.info("Executing activity 4")
        await workflow.execute_activity(my_activity_4)
        workflow.logger.info("Finished activity 4")

    @workflow.signal
    def start_activity(self, activity_number: int):
        if activity_number < 2 or activity_number > 4:
            raise ValueError("Activity number must be 2, 3, or 4")
        self.events[activity_number - 2].set()

This is a very normal workflow and will work just fine even if it has to wait for weeks between things, the worker crashes and it has to run on another worker, etc. Just send a signal with the activity number from a client anywhere in any language as needed. This is the magic of Temporal and our Temporal-backed asyncio event loop :slight_smile:

5

Thanks for the code @Chad_Retz! I think that i didn’t understand how Temporal Workflows working.