A modern and robust Python package for event-driven programming. Define, emit, and orchestrate events with ease using customizable event emitters and flexible responses.
Project description
Documentation: https://mdapena.github.io/pyventus
Source Code: https://github.com/mdapena/pyventus
Pyventus is a modern and robust Python package for event-driven programming. It offers a comprehensive suite of tools to define, emit, manage, and orchestrate events with ease, using customizable event emitters and flexible responses. With Pyventus, you can easily build scalable, extensible, and loosely-coupled event-driven applications.
More than just Events
Pyventus offers several key features that make it a powerful event-driven programming package for your Python projects:
- An Intuitive API ─ Pyventus provides a user-friendly API for defining events, emitters, and handlers. Its design simplifies the process of working with events, enabling you to organize your code around discrete events and their associated actions.
- Sync and Async Support ─ Whether your code is synchronous or asynchronous, Pyventus allows you to define event handlers as either sync or async callbacks and emit events from both scopes.
- Runtime Flexibility ─ Pyventus' runtime flexibility allows you to switch seamlessly between different official or custom event emitter implementations on the fly.
- Customization ─ Whether you choose official emitters or custom ones, Pyventus allows you to customize the behavior and capabilities of the event emitters to perfectly align with your unique requirements.
- Reliable Event Handling ─ Pyventus allows you to define handlers to customize how events are processed upon completion. Attach success and failure logic to take targeted actions based on the outcome of each event execution.
- Scalability and Maintainability ─ By adopting an event-driven approach with Pyventus, you can create scalable and maintainable code thanks to the loose coupling between its components that enables extensibility and modularity.
- Comprehensive Documentation ─ Pyventus provides a comprehensive documentation suite that includes API references, usage examples, best practices guides, and tutorials to effectively leverage all the features and capabilities of the package.
Requirements
Pyevents only requires Python 3.10+ by default, which includes the AsyncIOEventEmitter
and the ExecutorEventEmitter
with no additional dependencies. However, your requirements may expand
if you opt to use alternative built-in event emitter implementations.
Installation
Pyventus is available as a Python package and can be easily installed using pip
. Open your terminal
and execute the following command to install it:
pip install pyventus
Hello, World!
Example
Experience the power of Pyventus through a simple Hello, World!
example that illustrates
the core concepts and basic usage of the package. This example will walk you through the essential steps of
creating an event handler and triggering events within your application.
from pyventus import EventLinker, EventEmitter, AsyncIOEventEmitter
@EventLinker.on("MyEvent")
def event_callback():
print("Hello, World!")
event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("MyEvent")
You can also work with async
functions and contexts...
from pyventus import EventLinker, EventEmitter, AsyncIOEventEmitter
@EventLinker.on("MyEvent")
async def event_callback():
print("Hello, World!")
event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("MyEvent")
As we can see from the Hello, World!
example, Pyventus follows a simple and intuitive
workflow for event-driven programming. Let's recap the essential steps involved:
-
Defining the event handler callback: We defined the function
event_callback()
and used the@EventLinker.on()
decorator to associate it with the string eventMyEvent
. This decorator indicates that the function should be triggered whenMyEvent
occurs. -
Creating the event emitter instance: We instantiated the
AsyncIOEventEmitter
class, which acts as the event emitter responsible for dispatching events and invoking the associated event handler callbacks. -
Emitting the event: By utilizing the
emit()
method of the event emitter, we emitted the string eventMyEvent
. This action subsequently execute the registered event handler callbacks, which in this case is theevent_callback()
.
Having gained a clear understanding of the workflow showcased in the Hello, World!
example,
you are now well-equipped to explore more intricate event-driven scenarios and fully harness the capabilities of
Pyventus in your own projects.
Support for Synchronous and Asynchronous code
Pyventus is designed from the ground up to seamlessly support both synchronous and asynchronous
programming models. Its unified sync/async API allows you to define event handler callbacks and emit events
across sync
and async
contexts. Let's take a look at some use cases that illustrate
how the API handles event registration and dispatch transparently.
Registering Event Handlers as Sync
and Async
Callbacks
@EventLinker.on("MyEvent")
def sync_event_callback():
# Synchronous event handling
print("Event received!")
@EventLinker.on("MyEvent")
async def async_event_callback():
# Asynchronous event handling
print("Event received!")
Emitting Events from Sync
and Async
Contexts
# Emitting an event within a sync function
def sync_function(event_emitter: EventEmitter):
event_emitter.emit("MyEvent")
# Emitting an event within an async function
async def async_function(event_emitter: EventEmitter):
event_emitter.emit("MyEvent")
Event Propagation Within Different Contexts
While Pyventus provides a base EventEmitter
class with a unified sync/async API, the
specific propagation behavior when emitting events may vary depending on the concrete EventEmitter
used. For example, the AsyncioEventEmitter
implementation leverages the Asyncio
event
loop to schedule callbacks added from asynchronous contexts without blocking. But alternative emitters could
structure propagation differently to suit their needs.
Runtime Flexibility and Customization
At its core, Pyventus utilizes a modular event emitter design that allows you to switch seamlessly between different built-in or custom event emitter implementations on the fly. Whether you opt for official emitters or decide to create your custom ones, Pyventus allows you to tailor the behavior and capabilities of the event emitters to perfectly align with your unique requirements.
By leveraging the principles of dependency inversion and open-close, Pyventus decouples the event
emission process from the underlying implementation that handles the event handler callbacks and enables you,
in conjunction with the EventLinker
, to change the event emitter at runtime without the need to
reconfigure all connections or employ complex logic.
Seeing it in Action
Now let's put Pyventus' flexibility to the test with a practical example. First, we'll build a custom FastAPI-specific EventEmitter to properly handle the event handler callbacks using the framework's BackgroundTasks workflow. Then, we'll illustrate Pyventus' dynamic capabilities by swapping this custom emitter out for a built-in alternative on the fly.
To follow along, please ensure that you have the FastAPI framework installed. Once that is complete, let's dive into the step-by-step implementation:
-
Create a
main.py
file with:from asyncio import sleep from random import randint from typing import Any, List from fastapi import BackgroundTasks, FastAPI from pyventus import EventEmitter, EventHandler, EventLinker, AsyncIOEventEmitter class FastAPIEventEmitter(EventEmitter): """Custom event emitter class that leverages the FastAPI's asynchronous workflow.""" def __init__(self, background_tasks: BackgroundTasks): super().__init__() self._background_tasks = background_tasks # Store the FastAPI background tasks object def _execute(self, event_handlers: List[EventHandler], /, *args: Any, **kwargs: Any) -> None: for event_handler in event_handlers: # Execute event handlers as background tasks self._background_tasks.add_task(event_handler, *args, **kwargs) @EventLinker.on("console_print") async def console_print(message: str): await sleep(randint(0, 3)) # Simulate a random delay print(message) app = FastAPI() @app.get("/print") async def console_print_endpoint(background_tasks: BackgroundTasks): """FastAPI endpoint that triggers the console_print event.""" def console_print_app_service(event_emitter: EventEmitter) -> None: event_emitter.emit("console_print", message=f"\nHello, {event_emitter.__class__.__name__}!") # Emit the console_print event using FastAPIEventEmitter console_print_app_service(event_emitter=FastAPIEventEmitter(background_tasks)) # Emit the console_print event using AsyncIOEventEmitter console_print_app_service(event_emitter=AsyncIOEventEmitter()) return {"message": "Console print triggered!"}
-
Run the server with:
uvicorn main:app --reload
-
Open your browser at http://127.0.0.1:8000/print. You will
see the JSON response as:
{ "message": "Console print triggered!" }
You'll also be able to see the outputs of the event emitters in the console logs as:
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) INFO: Started reloader process [28720] INFO: Started server process [28722] INFO: Waiting for application startup. INFO: Application startup complete. INFO: 127.0.0.1:52391 - "GET /print HTTP/1.1" 200 OK Hello, FastAPIEventEmitter! Hello, AsyncIOEventEmitter!
As we can see from this practical example, we were able to easily adapt the event emitter to the context of the FastAPI framework. We defined and implemented a custom emitter tailored for FastAPI's workflow, using background tasks to handle the event handler callbacks accordingly.
We also saw Pyventus' dynamic flexibility firsthand - swapping emitters in real-time required no intricate reconfiguration or re-establishing of handlers. Simply changing the concrete emitter allowed for a seamless transition between implementations.
Defining Event Response Logic
As we mentioned earlier, Pyventus allows you to customize how events are processed upon completion in
success or error scenarios by attaching custom handlers. To utilize this functionality, Pyventus provides a simple
yet powerful Pythonic syntax through its event linkage context
.
The event linkage context
enables defining the event workflow and completion handlers in an
organized manner. This is done by using the EventLinker.on
method within a with
statement
block. Let's examine examples demonstrating how success and failure handlers can be attached using the event linkage
context:
Success and Error Handling Example
from pyventus import EventLinker, EventEmitter, AsyncIOEventEmitter
with EventLinker.on("StringEvent") as linker:
@linker.on_event
def event_callback() -> str:
print("Event received!")
return "Event succeeded!"
@linker.on_success
def success_callback(msg: str) -> None:
print(msg)
@linker.on_failure
def failure_callback(exc: Exception) -> None:
print(exc)
event_emitter: EventEmitter = AsyncIOEventEmitter()
event_emitter.emit("StringEvent")
As we have seen from the examples, Pyventus' event linkage context provides a reliable and Pythonic way to manage the event workflow and response to different completion outcomes through the use of custom handlers.
Continuous Evolution
Pyventus continuously adapts to support developers across technological and programming domains. Its aim is to remain at the forefront of event-driven design. Future development may introduce new official event emitters, expanding compatibility with different technologies through seamless integration.
Current default emitters provide reliable out-of-the-box capabilities for common use cases. They efficiently handle core event operations and lay the foundation for building event-driven applications.
Driving Innovation Through Collaboration
Pyventus is an open source project that welcomes community involvement. If you wish to contribute additional event emitters, improvements, or bug fixes, please check the Contributing section for guidelines on collaborating. Together, we can further the possibilities of event-driven development.
License
Pyventus is distributed as open source software and is released under the MIT License.
You can view the full text of the license in the LICENSE
file located in the Pyventus repository.
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.