How to dynamically add socketio event handler with a given string

Question

I have python-socketio server and client. When server calls client with event name, such as 'ping_client', client will emit back event 'ping_client_response' after receiving 'ping_client' event from the server.

Following server code works and print('got response from client') is called.

  sio = socketio.AsyncServer(async_mode='asgi')

  async def sio_request(event):

    @sio.event
    def ping_response(sid, file_pack):
      print('got response from client')

    await sio.emit(event, [])  


  sio_request('ping_client')

Now, I want to add asyncio event handler dynamically:

  sio = socketio.AsyncServer(async_mode='asgi')

  async def sio_request(event, handler_callback):

    # Define a function generator
    def create_event_handler(event_name):
        async def event_handler(sid, data):
            # Dynamic event handling logic
            handler_callback(sid, data)

        return event_handler

    # Create a dynamic event handler function
    dynamic_resp_handler = create_event_handler(event)

    func_name = event + '_response'
    # func_name = 'ping_client_response'
    setattr(sio, func_name, dynamic_resp_handler)

    await sio.emit(event, [])  

  sio_request('ping_client')

Client code:

    @sio.event
    def ping_client(data):
        print('ping_client from server')
        sio.emit('ping_client_response', 'ping_client_response from client')    

I am expecting handler_callback(sid, data) get called. But, it does not work.

My question is how to add python-socketio event handler dynamically from a given string?

Reference: https://python-socketio.readthedocs.io/en/stable/index.html

Answer 1

Adding event handlers while the server is running is not something that this library is designed for. You claim that your first example works, and I believe you, but this is not a great idea. Consider that there is no option to remove event handlers, so any handlers that you add in this way will stay in the system until you stop the server.

A better alternative in my opinion is to create a catch-all event handler that receives all events. You can then add your application logic to decide if an event is worth responding to or not:

@sio.on('*')
def catch_all(event, sid, data):
    # check event and decide if you want to handle it or not

But in any case, adding an event handler that has the name stored in a string is possible, in spite of not being very useful for the reasons stated above. You can do it like this:

sio.on(func_name)(dynamic_resp_handler)