Skip to content

The EventLinker Registry

🏗️ Work in Progress

This page is a work in progress.

  Events are essential for building reactive applications with Pyventus. However, we need a way to connect events to the code that should run in response. This is where the EventLinker comes in.

What is the EventLinker?

  The EventLinker is a central class that acts as a registry for linking events to their associated event handlers. It keeps track of which events have event handlers assigned to them, so when an event occurs it knows which code needs to run.

  The EventLinker can also be subclassed to create separate linking registries, allowing you to define different namespaces or contexts for events and event handlers.

Benefits of Using The EventLinker

Using the EventLinker class offers several advantages:

  • Clear separation of concerns - The emitter class focuses only on emitting events, while the linker handles all the subscription logic. This makes both classes cleaner and easier to understand, as well as allowing them to be modified independently as needed.
  • Centralized logic - Having global registries means there is only one place to go to see which events have event handlers. This simplifies management of the overall event system.
  • Flexibility - You can change the event emitter instance at runtime without the need to reconfigure all connections.

Event Handlers

  Before proceeding, we should first understand the concept behind event handlers and what they do. An event handler is a callable object designed to respond to a specific event. When an event occurs, such as a button click, mouse movement, or a timer expiration, these objects are executed concurrently in response.

  The EventHandler class encapsulates the event callbacks and provides a standardized mechanism for executing them when the event occurs. It handles the asynchronous and synchronous execution of event callbacks, as well as the completion workflow for success and error handling.

Subscribing Event Handlers

  The EventLinker makes it easy to subscribe event handlers to respond to events. Let's explore the different approaches to subscribing event handlers.

Subscription Basics

  Pyventus supports two types of subscriptions to handle callback registration: regular subscriptions and one-time subscriptions. Each subscription, regardless of type, will create a separate EventHandler instance independently. So subscribing the same callback multiple times to an event will cause it to be invoked multiple times.

Regular Subscriptions

  Regular subscriptions trigger the event handler each time the subscribed event(s) occur, and the handler remains subscribed until explicitly unsubscribed. Below we will explore how to subscribe regular event handlers.

Using decorators

Decorators provide a clean Python syntax for subscribing handlers. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# Subscribe to one event
@EventLinker.on("StringEvent")
def event_callback1():
    print("Event received!")


# Subscribe to multiple events at once
@EventLinker.on("StringEvent", "AnotherEvent", "ThirdEvent")
def event_callback2():
    print("Event received!")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# Subscribe to one event
@EventLinker.on("StringEvent")
async def event_callback1():
    print("Event received!")


# Subscribe to multiple events at once
@EventLinker.on("StringEvent", "AnotherEvent", "ThirdEvent")
async def event_callback2():
    print("Event received!")

Using the subscribe() method

You can also subscribe event handlers by calling the subscribe() method. 

1
2
3
4
5
6
7
8
9
def event_callback():
    print("Event received!")


# Subscribe to one event
EventLinker.subscribe("StringEvent", event_callback=event_callback)

# Subscribe to multiple events at once
EventLinker.subscribe("StringEvent", "AnotherEvent", "ThirdEvent", event_callback=event_callback)

1
2
3
4
5
6
7
8
9
async def event_callback():
    print("Event received!")


# Subscribe to one event
EventLinker.subscribe("StringEvent", event_callback=event_callback)

# Subscribe to multiple events at once
EventLinker.subscribe("StringEvent", "AnotherEvent", "ThirdEvent", event_callback=event_callback)

One-time Subscriptions

  One-time subscriptions trigger the event handler only once, then automatically unsubscribe it. One-time handlers are useful for tasks that should only run once.

Behavior with Multiple Events

When subscribing a one-time handler to multiple events, if one event fires it will automatically unsubscribe the event handler from all other events.

Using decorators

In order to perform a one-time subscription using decorators we use the once() method: 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# Subscribe to one event
@EventLinker.once("StringEvent")
def event_callback1():
    print("Event received!")


# Subscribe to multiple events at once
@EventLinker.once("StringEvent", "AnotherEvent", "ThirdEvent")
def event_callback2():
    print("Event received!")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# Subscribe to one event
@EventLinker.once("StringEvent")
async def event_callback1():
    print("Event received!")


# Subscribe to multiple events at once
@EventLinker.once("StringEvent", "AnotherEvent", "ThirdEvent")
async def event_callback2():
    print("Event received!")

Using the subscribe() method

Alternatively, you can also use the subscribe() method to do a one-time subscription too: 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
def event_callback():
    print("Event received!")


# Subscribe to one event
EventLinker.subscribe(
    "StringEvent", 
    event_callback=event_callback, 
    once=True,
)

# Subscribe to multiple events at once
EventLinker.subscribe(
    "StringEvent", "AnotherEvent", "ThirdEvent", 
    event_callback=event_callback, 
    once=True,
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
async def event_callback():
    print("Event received!")


# Subscribe to one event
EventLinker.subscribe(
    "StringEvent", 
    event_callback=event_callback, 
    once=True,
)

# Subscribe to multiple events at once
EventLinker.subscribe(
    "StringEvent", "AnotherEvent", "ThirdEvent", 
    event_callback=event_callback, 
    once=True,
)

Success and Error Handling

  In the previous sections, we discussed the process of subscribing callback functions to handle events using Pyventus' event linker. However, there may be times when we need more control over the exact workflow.

  In this section, we'll show you how to define custom logic to process events upon completion using success and failure handlers. It's important to have reliable control over the asynchronous flow to build robust apps.

Response Logic For Regular Subscriptions

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
with EventLinker.on("DivisionEvent") as linker:  # (1)!

    @linker.on_event
    def handle_division(a: float, b: float) -> float:
        return a / b

    @linker.on_success
    def handle_success(result: float) -> None:
        print(f"Division result: {result:.3g}")

    @linker.on_failure
    def handle_failure(e: Exception) -> None:
        print(f"Oops, something went wrong: {e}")
  1. When the EventLinker.on method is used as a context manager via the with statement, it allows multiple callbacks to be associated with events within the linkage context block, defining the event workflow.
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
def handle_division(a: float, b: float) -> float:
    return a / b

def handle_success(result: float) -> None:
    print(f"Division result: {result:.3g}")

def handle_failure(e: Exception) -> None:
    print(f"Oops, something went wrong: {e}")


EventLinker.subscribe(
    "DivisionEvent", 
    event_callback=handle_division, 
    success_callback=handle_success, 
    failure_callback=handle_failure,
)

Response Logic For One-time Subscriptions

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
with EventLinker.once("DivisionEvent") as linker:  # (1)!

    @linker.on_event
    def handle_division(a: float, b: float) -> float:
        return a / b

    @linker.on_success
    def handle_success(result: float) -> None:
        print(f"Division result: {result:.3g}")

    @linker.on_failure
    def handle_failure(e: Exception) -> None:
        print(f"Oops, something went wrong: {e}")
  1. When the EventLinker.once method is used as a context manager via the with statement, it allows multiple callbacks to be associated with events within the linkage context block, defining the event workflow.
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
def handle_division(a: float, b: float) -> float:
    return a / b

def handle_success(result: float) -> None:
    print(f"Division result: {result:.3g}")

def handle_failure(e: Exception) -> None:
    print(f"Oops, something went wrong: {e}")


EventLinker.subscribe(
    "DivisionEvent", 
    event_callback=handle_division, 
    success_callback=handle_success, 
    failure_callback=handle_failure,
    once=True,
)

Optimizing Callbacks Execution

  By default, event handlers in Pyventus are executed concurrently during an event emission, running their sync and async callbacks as defined. However, if you have a sync callback that involves I/O or non-CPU bound operations, you can enable the force_async parameter to offload it to a thread pool, ensuring optimal performance and responsiveness. The force_async parameter utilizes the asyncio.to_thread() function to execute sync callbacks asynchronously. 

1
2
3
4
5
6
7
8
9
# You can also set this when using the `with` 
# statement and the `once()` decorator
@EventLinker.on("BlockingIO", force_async=True)
def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    # Note that time.sleep() can be replaced with any blocking
    # IO-bound operation, such as file operations.
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    # Note that time.sleep() can be replaced with any blocking
    # IO-bound operation, such as file operations.
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

EventLinker.subscribe(
    "BlockingIO",
    event_callback=blocking_io,
    force_async=True,
)

Unsubscribing Event Handlers

  Removing event handlers is an important part of working with events. Let's look at different approaches to properly teardown event subscriptions:

Removing an Event Handler from an Event

To remove a single event handler from a specific event: 

1
2
3
4
5
6
7
def event_callback():
    print("Event received!")


event_handler = EventLinker.subscribe("StringEvent", "AnotherEvent", event_callback=event_callback)

EventLinker.unsubscribe("StringEvent", event_handler=event_handler)  # (1)!
  1. Removes the event_handler from just the StringEvent

Removing Event Handlers from All Events

To remove an event handler from all subscribed events: 

1
2
3
4
5
6
7
def event_callback():
    print("Event received!")


event_handler = EventLinker.subscribe("StringEvent", "AnotherEvent", event_callback=event_callback)

EventLinker.remove_event_handler(event_handler=event_handler)  # (1)!
  1. Removes the event_handler from the StringEvent and AnotherEvent

Removing an Event and its Event Handlers

To delete an event and all associated handlers: 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
@EventLinker.on("StringEvent")
def event_callback1():
    print("Event received!")


@EventLinker.on("StringEvent", "AnotherEvent")
def event_callback2():
    print("Event received!")


EventLinker.remove_event(event="StringEvent")  # (1)!
  1. Removes all event handlers associated with the StringEvent

Clearing All Events

To remove all events and their handlers: 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
@EventLinker.on("StringEvent", ...)
def event_callback1():
    pass


@EventLinker.on(...)
def event_callback2():
    pass


EventLinker.remove_all()

Custom Event Linkers

  The EventLinker class in Pyventus was designed to support subclassing to allow you to define separate linking registries or namespaces for your events and handlers, as well as configure the EventLinker behavior. This approach provides a powerful way to customize how events are linked within your applications. Some key reasons for using custom linkers include:

  • Organizing events into logical domains/contexts.
  • Isolating events to only be accessible within certain scopes.
  • Configuring linker-specific options like max handlers per event.

To define a custom linker, subclass the EventLinker, as shown below: 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
class UserEventLinker(EventLinker, max_event_handlers=10):
    """ EventLinker for User's events only """
    pass  # Additional logic can be added here if needed...


@UserEventLinker.on("PasswordResetEvent")
async def handle_password_reset_event(email: str):
    print("PasswordResetEvent received!")


@UserEventLinker.on("EmailVerifiedEvent")
async def handle_email_verified_event(email: str):
    print("EmailVerifiedEvent received!")

Debug Mode

  The EventLinker also offers a debug mode feature which helps you understand how event subscriptions and unsubscriptions are happening during runtime.

Global Debug Mode

  By default, Pyventus leverages the Python's global debug tracing feature. Simply run your code in an IDE's debugger mode to activate the global debug mode tracing.

EventLinker Global Debug Mode

Namespace Debug Flag

  Alternatively, if you want to enable or disable the debug mode specifically for a certain EventLinker namespace, you can use the debug flag that is available in the subclass configurations. Setting the debug flag to True enables debug mode for that namespace, while setting it to False disables debug mode. Here's an example: 

1
2
class CustomEventLinker(EventLinker, debug=True):
    pass  # Additional logic can be added here if needed...

1
2
class CustomEventLinker(EventLinker, debug=False):
    pass  # Additional logic can be added here if needed...

Recap

In this tutorial we covered:

  • The standardized mechanism for executing event callbacks (Event Handlers).
  • Subscribing regular and one-time handlers with decorators and the subscribe() method.
  • Unsubscribing a single event handler, all handlers, an event, or clearing all.
  • Success and error handling by defining custom logic for event completion.
  • Techniques for optimizing synchronous callback execution.
  • Custom Event Linkers to separate event namespaces.
  • Debug mode to trace subscriptions

We learned the core EventLinker concepts of:

  • Use the EventLinker to link events to code responses.
  • Subscribe/unsubscribe as needed using various approaches.