API Reference

class wsrpc_aiohttp.WSRPCBase(loop: asyncio.events.AbstractEventLoop = None)[source]

Common WSRPC abstraction

classmethod add_route(route: str, handler: typing.Union[wsrpc_aiohttp.websocket.route.WebSocketRoute, typing.Callable])[source]

Expose local function through RPC

  • route – Name which function will be aliased for this function. Remote side should call function by this name.
  • handler – Function or Route class (classes based on wsrpc_aiohttp.WebSocketRoute). For route classes the public methods will be registered automatically.


Route classes might be initialized only once for the each socket instance.

In case the method of class will be called first, wsrpc_aiohttp.WebSocketRoute.init() will be called without params before callable method.

call(func: str, **kwargs)[source]

Method for call remote function

Remote methods allows only kwargs as arguments.

You might use functions as route or classes

async def remote_function(socket: WSRPCBase, *, foo, bar):
    # call function from the client-side
    await self.socket.proxy.ping()
    return foo + bar

class RemoteClass(WebSocketRoute):

    # this method executes when remote side call route name
    asyc def init(self):
        # call function from the client-side
        await self.socket.proxy.ping()

    async def make_something(self, foo, bar):
        return foo + bar

Property which contains the socket clients


Cancel all pending tasks


Special property which allow run the remote functions by dot notation

# calls remote function with name ping
await client.proxy.ping()

# full equivalent of
await client.call('ping')
classmethod remove_route(route: str, fail=True)[source]

Removes route by name. If fail=True an exception will be raised in case the route was not found.


Property which contains the socket routes

class wsrpc_aiohttp.WebSocketBase(request)[source]

Base class for aiohttp websocket handler

authorize() → bool[source]

Special method for authorize client. If this method return True then access allowed, otherwise 403 Forbidden will be sent.

This method will be called before socket connection establishment.

By default everyone has access. You have to inherit this class and change this behaviour.


You can validate some headers (self.request.headers) or check cookies (self.reauest.cookies).

classmethod broadcast(func, callback=<bound method WebSocketRoute.placebo of <class 'wsrpc_aiohttp.websocket.route.WebSocketRoute'>>, **kwargs)[source]

Call remote function on all connected clients

  • func – Remote route name
  • callback – Function which receive responses

Cancel all pending tasks and stop this socket connection

classmethod configure(keepalive_timeout=30, client_timeout=10, max_concurrent_requests=25)[source]

Configures the handler class

  • keepalive_timeout – sets timeout of client pong response
  • client_timeout – internal lock timeout
  • max_concurrent_requests – how many concurrent requests might be performed by each client
class wsrpc_aiohttp.WebSocketAsync(request)[source]

Handler class which execute any route as a coroutine

class wsrpc_aiohttp.WebSocketThreaded(request)[source]

Handler class which execute any route in the default thread-pool of current event loop

class wsrpc_aiohttp.WSRPCClient(endpoint: typing.Union[yarl.URL, str], loop=None, timeout=None, session: aiohttp.client.ClientSession = None, **kwargs)[source]

WSRPC Client class


Close the client connect connection


Perform connection to the server


singledispatch wrapped function. You might register custom types if you want pass it to the remote side.

from wsrpc_aiohttp import serializer

class MyObject:
    def __init__(self):
        self.foo = 'bar'

def _(value: MyObject) -> dict:
    return {'myObject': {'foo': value.foo}}