How do I create an event bus using python?

Question

So, I have a script I made that tracks twitter follows. The bulk of the script's work is done in two parts. One function track_user(old_followers: dict, screen_name: str) -> dict, where dict contains information about new followers and unfollowers. The second function takes the dict from that first function, parses it, and converts it to text that makes sense.

My predicament is: I want other users of this script to be able to insert their own separate from my own code.

I'll give an example below

# My code

def track_user(old_followers: dict, screen_name: str) -> dict:
    final_dict = tracking_code_here()
    SomeEventBus.emit(TrackUserEvent(final_dict))
    return final_dict

# Some user's code (using my example above as an example)
def convert_tracking_to_text(event: TrackUserEvent): 
    perform_some_text_conversion(event.final_dict)
    return

# Maybe a decorator can be used in this case instead
MyScript.EventBus.AddListener(TrackUserEvent, convert_tracking_to_text)

Hopefully, this makes sense. I just want my script to emit events that others can import and listen to.

Hervé · Accepted Answer · 2022-06-01 13:43:06Z

1

I made some time ago a library for handling async events. Basically you could do things such as

# creates a main event manager
manager = EventManager()

# creates an event
track_user_event = manager.create_event("track_user_event")

async def track_user(old_followers: dict, screen_name: str) -> dict:
    final_dict = tracking_code_here()
    
    # emit event here
    await track_user_event(TrackUserEvent(finalDict)) # could also be await manager.raise_event("track_user_event");

    return final_dict

@track_user_event.as_callback()
async def convert_tracking_to_text(event: TrackUserEvent): 
    perform_some_text_conversion(event.final_dict)
    return

It's far away to be perfect but I hope it could help you.

answered Jun 1, 2022 at 13:43

Hervé

1,0187 silver badges13 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

9 Guy Over a year ago

Thank you for linking your library. I ended up going with a smaller library, but yours would've worked well for my use case as well. It has a lot more options which I appreciate. I'll keep it in mind for future purposes

9 Guy · Accepted Answer · 2023-03-16 18:59:55Z

I accepted the other answer because it was useful and did answer the question, but I used the event-bus library instead.

e.g

def do_query():
    # Announce that user query starting
    MAIN_BUS.emit(events.StartQueryEvent.EVENT_NAME)
    query_users_events = []

    for usr in USERS_TO_QUERY:
        this_user = some_query(usr)
        # Emit event for this specific user
        MAIN_BUS.emit(events.TwitterUserFollowingQuery.EVENT_NAME, this_user)

        # Keeping track of all queried users
        query_users_events.append(this_user)

    # Emit an event with data of all queried users
    MAIN_BUS.emit(events.FinishAllQueryEvents.EVENT_NAME, query_users_events)

Then in another file,

# Compare old followers/mutuals to current mutuals after all query events are completed
@bus.on(events.FinishAllQueryEvents.EVENT_NAME)
def keep_track(user_data: list[events.TwitterUserFollowingQuery]):
    some_logic_based_on_data(user_data)

Full example repo

Collectives™ on Stack Overflow

How do I create an event bus using python?

2 Answers 2

1 Comment

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

1 Comment

Comments

Your Answer

Sign up or log in

Post as a guest

Related