aiozmq does not work when using type hinting
ithinuel opened this issue · 2 comments
ithinuel commented
Parameter checking fails when usign type hinting with custom types.
import asyncio
import aiozmq, aiozmq.rpc
from collections import namedtuple
import msgpack
class Point:
def __init__(self, x, y):
self.x = x
self.y = y
def __eq__(self, other):
if isinstance(other, Point):
return (self.x, self.y) == (other.x, other.y)
return NotImplemented
ATuple = namedtuple('ATuple', "a, b")
AnotherTuple = namedtuple('AnotherTuple', "c, d")
translation_table = {
0: (Point,
lambda value: msgpack.packb((value.x, value.y)),
lambda binary: Point(*msgpack.unpackb(binary))),
1: (ATuple,
lambda value: msgpack.packb((value.a, value.b)),
lambda binary: ATuple(*msgpack.unpackb(binary)))
}
class ServerHandler(aiozmq.rpc.AttrHandler):
@aiozmq.rpc.method
async def remote(self, val):
return val
@aiozmq.rpc.method
async def remote_with_hint(self, val: Point) -> Point:
return val
@aiozmq.rpc.method
async def atuple(self, val):
return val
@aiozmq.rpc.method
async def atuple_with_hint(self, val: ATuple) -> ATuple:
return val
@aiozmq.rpc.method
async def another_tuple(self, val):
return val
@aiozmq.rpc.method
async def another_tuple_with_hint(self, val: AnotherTuple) -> AnotherTuple:
return val
async def go():
server = await aiozmq.rpc.serve_rpc(
ServerHandler(), bind='tcp://127.0.0.1:5555',
translation_table=translation_table
)
client = await aiozmq.rpc.connect_rpc(
connect='tcp://127.0.0.1:5555',
translation_table=translation_table
)
ret = await client.call.remote(Point(1, 2))
assert ret == Point(1, 2)
ret = await client.call.remote_with_hint(Point(1, 2))
assert ret == Point(1, 2)
asyncio.get_event_loop().run_until_complete(go())
ShT3ch commented
I'm having problems with type annotation too.
I use trafaret, but it works poor with tuples(lists converted to tuples during message passing)
Here is issue I created in trafaret's repo.
It seems that this issue should be addressed there.