Source code for wsrpc_aiohttp.websocket.tools
import asyncio
import base64
from functools import singledispatch, wraps
class Lazy:
__slots__ = ("func",)
def __init__(self, func):
self.func = func
def __str__(self):
return str(self.func())
def __repr__(self):
return repr(self.func())
[docs]
@singledispatch
def serializer(value):
""" singledispatch wrapped function.
You might register custom types if you want pass it to the remote side.
.. code-block:: python
from wsrpc_aiohttp import serializer
class MyObject:
def __init__(self):
self.foo = 'bar'
@serializer.register(MyObject)
def _(value: MyObject) -> dict:
return {'myObject': {'foo': value.foo}}
"""
raise ValueError("Can not serialize %r" % type(value))
@serializer.register(bytes) # noqa: W0404
def _(value):
return base64.b64encode(value).decode()
class SingletonMeta(type):
def __new__(cls, clsname, superclasses, attributedict):
klass = type.__new__(cls, clsname, superclasses, attributedict)
klass.__instance__ = None
return klass
class Singleton(metaclass=SingletonMeta):
def __new__(cls, *args, **kwargs):
if not cls.__instance__:
cls.__instance__ = super(Singleton, cls).__new__(cls)
return cls.__instance__
def awaitable(func):
if asyncio.iscoroutinefunction(func):
return func
@wraps(func)
async def wrap(*args, **kwargs):
result = func(*args, **kwargs)
is_awaitable = (
asyncio.iscoroutine(result)
or asyncio.isfuture(result)
or hasattr(result, "__await__")
)
if is_awaitable:
return await result
return result
return wrap
__all__ = (
"Lazy",
"Singleton",
"awaitable",
"serializer",
)