drive-events 0.0.1.dev4

Build event-driven workflows with python functions

drive-events

Build event-driven workflows with python async functions

Install

Install from PyPi

pip install drive-events

Install from source

# clone this repo first
cd drive-events
pip install -e .

Quick Start

A hello world example:

import asyncio
from drive_events import EventInput, default_drive


@default_drive.make_event
async def hello(event: EventInput, global_ctx):
    print("hello")


@default_drive.listen_groups([hello])
async def world(event: EventInput, global_ctx):
    print("world")

asyncio.run(default_drive.invoke_event(hello))

In this example, The return of hello event will trigger world event.

To make an event function, there are few elements:

• Input Signature: must be (event: EventInput, global_ctx)
  • EventInput is the returns of the listening groups.
  • global_ctx is set by you when invoking events, it can be anything and default to None
• Make sure you decorate the function with @default_drive.make_event or @default_drive.listen_groups([EVENT,...])

Then, run your workflow from any event:

await default_drive.invoke_event(EVENT, EVENT_INPUT, GLOBAL_CTX)

Check out examples for more user cases!