'Syntax enforceability in python event queue

I wrote a multi-producer single consumer system where Service consumes events that arrive from different threads and processes them. Here's how the consumer looks:

@dataclass
class RPCEvent:
    method: str
    args: Tuple[Any] = field(default_factory=tuple)
    kwargs: Dict[Any, Any] = field(default_factory=dict)

RPCEventQueue = Queue[RPCEvent] 

class Service:
    def __init__(self):
        self.queue = RPCEventQueue()
    
    def fun(self, message: str):
        print(f"fun({message})")

    def run(self):
        while True:
            try:
                event = self.queue.get(1)
                getattr(event.method)(*event.args, **event.kwargs)
            except Empty:
                continue
            except Exception:
                break
            finally:
                self.queue.task_done()

     # from another thread I push events into Service's queue
     # service.queue.put_nowait(RPCEvent(method="fun", kwargs={"message": "one"}))

This works great, however, there is no syntax enforceability between RPCEvent and Service. For example I could do this:

service.queue.put_nowait(RPCEvent(method="fun", kwargs={"oops": "oh no"}))

Is there a way of knowing that Service.fun has no oops parameter at the time of writing this line?

I checked Protocol docs, but I couldn't find a creative way to make it suit my case.



Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source