pyventus 0.4.0

A modern and robust Python package for event-driven programming. Define, emit, and orchestrate events with ease using customizable event emitters and flexible responses.

---

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:

1. Defining the event handler callback: We defined the function event_callback() and used the @EventLinker.on() decorator to associate it with the string event MyEvent. This decorator indicates that the function should be triggered when MyEvent occurs.
2. 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.
3. Emitting the event: By utilizing the emit() method of the event emitter, we emitted the string event MyEvent. This action subsequently execute the registered event handler callbacks, which in this case is the event_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 emission 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 EventEmitter to properly handle the event emission 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:

1. Create a main.py file with:

   from asyncio import sleep
   from random import randint
   
   from fastapi import BackgroundTasks, FastAPI
   
   from pyventus import EventEmitter, 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 _process(self, event_emission: EventEmitter.EventEmission) -> None:
           self._background_tasks.add_task(event_emission)  # Execute the event emission as background tasks
   
   
   @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!"}
   

2. Run the server with:

   uvicorn main:app --reload
   

3. 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 emission 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.

Official FastAPIEventEmitter integration

No need for manual implementation! Pyventus now offers an official FastAPIEventEmitter integration.

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.