'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 |
|---|
