Supporting multiple event names per event

[0x-logic]

Currently, an event maps to a single event name via get_valid_event_name().

It would be nice if an event could map to a set of names, linking to subscribers that handle any one of the names.

Concretely, this would unlock subscribing to an event's inheritance structure.

import inspect


class EventLinker:
    @classmethod
    def get_valid_event_names(cls, event: SubscribableEventType) -> set[str]:
        # If `event` is a type, then walk the MRO for all of the parent
        # types, excluding `object`. Return the set of their names.
        if isinstance(event, type):
            return {t.__name__ for t in inspect.getmro(tp) if t is not object}

        # If the event is not supported, raise an exception
        raise PyventusException("Unsupported event")  # noqa: EM101 TRY003

For instance:

from pydantic import BaseModel

from typing import Generic, TypeVar


T = TypeVar("T")

# Example types show diamond inheritance for completeness

class A(BaseModel, Generic[T]):
    ttt: T

class B(A): ...

class C(A): ...

class D(B, C): ...

assert EventLinker.get_valid_event_names(D) == {'Generic', 'B', 'D', 'A', 'BaseModel', 'C'}

I don't think the change would be too disruptive in EventLinker. There might be a little ugliness in shimming around the existing get_valid_event_name() pattern to avoid a breaking change on end-user overrides (especially given the current suggestion for Pydantic handling).

The benefits would be pretty huge though, since you could now design a tree of domain-specif Pydantic model event classes, and choose your event subscription at any level of the tree, from the most specific events to the most generic, depending upon your subscription needs.

[0x-logic]

bump?

[mdapena]

Hey @0x-logic!

Thanks for taking the time to share your ideas!

Believe it or not, this feature was actually part of the initial proposal for the Pyventus subscription model. I wanted to mimic the Wildcard feature that many event emitter libraries offer, but instead of using strings, I thought about using the inheritance tree. However, after trying it out a few times, I realized that it could be more problematic than helpful. When someone subscribes to a simple class, they might not know all the superclasses attached unless they’re familiar with the inheritance structure. This could lead to unexpected behavior, like emitting a random event that triggers subscribers you didn’t anticipate.

That said, this idea inspired the current subscription model, which allows you to subscribe to any number of events at once by separating them with commas. This approach not only gives you the flexibility to specify exactly which events you want to attach to the subscriber, but it also makes it visually clear since you have to list all the classes and superclasses involved. It’s a way to keep everything organized and understandable.

If you’re looking to implement this functionality automatically, one idea I had is to create a helper function that returns the inheritance tree of a given class. This way, when you use the subscribe method of the event linker, you can just unpack the inheritance tree without needing to modify the get_valid_event_names() method. Here’s a quick code example to illustrate this:

from dataclasses import dataclass, is_dataclass
from typing import Generic, TypeVar

from pyventus.events import EventLinker

# Define a type variable for generic class usage
T = TypeVar("T")


@dataclass
class A(Generic[T]):
    ttt: T


@dataclass
class B(A):
    pass


@dataclass
class C(A):
    pass


@dataclass
class D(B, C):
    pass


def inheritance_tree(cls: type) -> list[type]:
    """
    Returns a list of superclasses in the inheritance hierarchy
    of the given class.
    """
    return [
        clstype
        for clstype in cls.__mro__
        if is_dataclass(clstype) or issubclass(clstype, Exception)
        # You can modify this condition as needed to include other desired classes in the inheritance structure.
    ]


# Subscribe to events from the inheritance tree of class D
subscriber = EventLinker.subscribe(*inheritance_tree(D), event_callback=print)

# Print the events associated with the subscribers
print(EventLinker.get_events_from_subscribers(subscriber))

Doing it this way keeps things clear, and the method name for the inheritance_tree() makes it obvious that you’re subscribing to the inheritance tree instead of just the simple class, which is a good reminder to be cautious. For now, though, this feature won’t be part of the main Pyventus library since it can be easily implemented through a function and is more of a user-specific need. But don’t get me wrong, if there’s enough demand from users, it could definitely be considered for future updates!

I hope this helps with your specific use case! If you have any questions or need more clarification, just let me know.

Happy coding!

[0x-logic]

Ah, interesting.
Ya, that makes sense about it possibly being a surprising footgun. Especially if the base type is a commonly-used built-in.

The inheritance_tree approach could work for some cases, but is a little opposite how I'd imagine would be useful in our case.

In our mind it would be about subscribing to multiple (all of the subclasses) in order to catch events broadly, without having to keep subscription list up to date.

ie,

@dataclass
class DatabaseEvent:
    pass

@dataclass
class PostgresEvent(DatabaseEvent):
    pass

@dataclass
class MongoEvent(DatabaseEvent):
    pass


@dataclass
class InsertEvent(DatabaseEvent):
    pass

@dataclass
class UpdateEvent(DatabaseEvent):
    pass


@dataclass
class PostgresInsertEvent(PostgresEvent, InsertEvent):
    pass

@dataclass
class PostgresUpdateEvent(PostgresEvent, UpdateEvent):
    pass


@dataclass
class MongoInsertEvent(MongoEvent, InsertEvent):
    pass

@dataclass
class MongoUpdateEvent(MongoEvent, UpdateEvent):
    pass


# Subscribe to any DB event
EventLinker.subscribe(DatabaseEvent, event_callback=lambda e: print(e))

# Subscribe to any Postgres event
EventLinker.subscribe(PostgresEvent, event_callback=lambda e: print(e))

# Subscribe to any Insert event
EventLinker.subscribe(InsertEvent, event_callback=lambda e: print(e))

Obviously contrived, but gives you an idea of how a subclass handling approach could be used to more generally handle custom event type trees. Ie, making use of diamond or multi-inheritance to create a domain matrix.