wsrpc-asgi API¶

Signatures from version 0.1.0 (source revision 7d95443bc646d77dbf4df2013a0faf3ea1322606).

This reference is checked in so builds do not require installing the library or accessing its source repository.

WSRPCBase¶

class wsrpc_asgi.WSRPCBase(loop: asyncio.AbstractEventLoop | None = None, timeout: TimeoutType | None = None, loads: LoadsType = json.loads, dumps: DumpsType = json.dumps)¶

Common WSRPC abstraction

async close(message=None)¶: Fail pending calls, stop tasks and await route cleanup once.

async handle_binary(message: bytes)¶: Override to handle binary WebSocket messages.

async handle_message(message: str)¶

classmethod get_routes()¶

classmethod get_clients()¶

property routes¶: Property which contains the socket routes

property clients¶: Property which contains the socket clients

prepare_args(args)¶

static is_route(func)¶

async handle_method(method, serial, args, kwargs)¶

async handle_result(serial, result)¶

async handle_error(serial, error)¶

async handle_event(event)¶

resolver(func_name)¶

async call(func: str, timeout=None, **kwargs)¶

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 socket.proxy.ping()
    return foo + bar

class RemoteClass(WebSocketRoute):

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

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

async emit(event)¶

classmethod add_route(route: str, handler: RouteType)¶

Expose local function through RPC

Parameters:

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_asgi.WebSocketRoute). For route classes the public methods will be registered automatically.

Note

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

A route instance is created lazily for each connection. Calling the bare route name resolves its exposed init method; init is not called automatically before other methods.

add_event_listener(func: EventListenerType)¶

remove_event_listeners(func)¶

classmethod remove_route(route: str, fail=True)¶: Removes route by name. If fail=True an exception will be raised in case the route was not found.

property proxy¶

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

WSRPCClient¶

class wsrpc_asgi.WSRPCClient(endpoint: str, loop: asyncio.AbstractEventLoop | None = None, timeout: TimeoutType | None = None, loads: LoadsType = json.loads, dumps: DumpsType = json.dumps, **kwargs: Any)¶

Bidirectional RPC client backed by websockets.asyncio.

Pass a ws:// or wss:// URL. Extra keyword arguments go to websockets.asyncio.client.connect.

async close(message=None)¶

async handle_binary(message: bytes)¶: Override to handle binary WebSocket messages.

async handle_message(message: str)¶

classmethod get_routes()¶

classmethod get_clients()¶

property routes¶: Property which contains the socket routes

property clients¶: Property which contains the socket clients

prepare_args(args)¶

static is_route(func)¶

async handle_method(method, serial, args, kwargs)¶

async handle_result(serial, result)¶

async handle_error(serial, error)¶

async handle_event(event)¶

resolver(func_name)¶

async call(func: str, timeout=None, **kwargs)¶

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 socket.proxy.ping()
    return foo + bar

class RemoteClass(WebSocketRoute):

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

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

async emit(event)¶

classmethod add_route(route: str, handler: RouteType)¶

Expose local function through RPC

Parameters:

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_asgi.WebSocketRoute). For route classes the public methods will be registered automatically.

Note

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

A route instance is created lazily for each connection. Calling the bare route name resolves its exposed init method; init is not called automatically before other methods.

add_event_listener(func: EventListenerType)¶

remove_event_listeners(func)¶

classmethod remove_route(route: str, fail=True)¶: Removes route by name. If fail=True an exception will be raised in case the route was not found.

property proxy¶

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

async connect()¶

WebSocketBase¶

class wsrpc_asgi.WebSocketBase(socket: WebSocketLike, *, context: Any = None)¶

One RPC session. Subclass to customize authorization and hooks.

async close(message=None)¶

async handle_binary(message: bytes)¶: Override to handle binary WebSocket messages.

async handle_message(message: str)¶

classmethod get_routes()¶

classmethod get_clients()¶

property routes¶: Property which contains the socket routes

property clients¶: Property which contains the socket clients

prepare_args(args)¶

static is_route(func)¶

async handle_method(method, serial, args, kwargs)¶

async handle_result(serial, result)¶

async handle_error(serial, error)¶

async handle_event(event)¶

resolver(func_name)¶

async call(func: str, timeout=None, **kwargs)¶

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 socket.proxy.ping()
    return foo + bar

class RemoteClass(WebSocketRoute):

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

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

async emit(event)¶

classmethod add_route(route: str, handler: RouteType)¶

Expose local function through RPC

Parameters:

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_asgi.WebSocketRoute). For route classes the public methods will be registered automatically.

Note

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

A route instance is created lazily for each connection. Calling the bare route name resolves its exposed init method; init is not called automatically before other methods.

add_event_listener(func: EventListenerType)¶

remove_event_listeners(func)¶

classmethod remove_route(route: str, fail=True)¶: Removes route by name. If fail=True an exception will be raised in case the route was not found.

property proxy¶

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 configure(keepalive_timeout=30, client_timeout=10, max_concurrent_requests=25, loads=json.loads, dumps=json.dumps)¶

classmethod freeze()¶

async authorize()¶: Return False to reject the WebSocket handshake with HTTP 403.

async classmethod serve(socket: WebSocketLike, *, context: Any = None)¶: Accept and serve a Starlette/FastAPI WebSocket until disconnect.

classmethod as_asgi(*, path: str | None = None)¶: Create an isolated ASGI endpoint with an optional path.

async run()¶

classmethod broadcast(func, callback=None, return_exceptions=True, **kwargs)¶

WebSocketAsync¶

class wsrpc_asgi.WebSocketAsync(socket: WebSocketLike, *, context: Any = None)¶

Execute synchronous and asynchronous handlers on the event loop.

async close(message=None)¶

async handle_binary(message: bytes)¶: Override to handle binary WebSocket messages.

async handle_message(message: str)¶

classmethod get_routes()¶

classmethod get_clients()¶

property routes¶: Property which contains the socket routes

property clients¶: Property which contains the socket clients

prepare_args(args)¶

static is_route(func)¶

async handle_method(method, serial, args, kwargs)¶

async handle_result(serial, result)¶

async handle_error(serial, error)¶

async handle_event(event)¶

resolver(func_name)¶

async call(func: str, timeout=None, **kwargs)¶

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 socket.proxy.ping()
    return foo + bar

class RemoteClass(WebSocketRoute):

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

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

async emit(event)¶

classmethod add_route(route: str, handler: RouteType)¶

Expose local function through RPC

Parameters:

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_asgi.WebSocketRoute). For route classes the public methods will be registered automatically.

Note

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

A route instance is created lazily for each connection. Calling the bare route name resolves its exposed init method; init is not called automatically before other methods.

add_event_listener(func: EventListenerType)¶

remove_event_listeners(func)¶

classmethod remove_route(route: str, fail=True)¶: Removes route by name. If fail=True an exception will be raised in case the route was not found.

property proxy¶

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 configure(keepalive_timeout=30, client_timeout=10, max_concurrent_requests=25, loads=json.loads, dumps=json.dumps)¶

classmethod freeze()¶

async authorize()¶: Return False to reject the WebSocket handshake with HTTP 403.

async classmethod serve(socket: WebSocketLike, *, context: Any = None)¶: Accept and serve a Starlette/FastAPI WebSocket until disconnect.

classmethod as_asgi(*, path: str | None = None)¶: Create an isolated ASGI endpoint with an optional path.

async run()¶

classmethod broadcast(func, callback=None, return_exceptions=True, **kwargs)¶

WebSocketThreaded¶

class wsrpc_asgi.WebSocketThreaded(socket: WebSocketLike, *, context: Any = None)¶

Run synchronous handlers in a thread and await async handlers.

async close(message=None)¶

async handle_binary(message: bytes)¶: Override to handle binary WebSocket messages.

async handle_message(message: str)¶

classmethod get_routes()¶

classmethod get_clients()¶

property routes¶: Property which contains the socket routes

property clients¶: Property which contains the socket clients

prepare_args(args)¶

static is_route(func)¶

async handle_method(method, serial, args, kwargs)¶

async handle_result(serial, result)¶

async handle_error(serial, error)¶

async handle_event(event)¶

resolver(func_name)¶

async call(func: str, timeout=None, **kwargs)¶

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 socket.proxy.ping()
    return foo + bar

class RemoteClass(WebSocketRoute):

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

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

async emit(event)¶

classmethod add_route(route: str, handler: RouteType)¶

Expose local function through RPC

Parameters:

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_asgi.WebSocketRoute). For route classes the public methods will be registered automatically.

Note

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

A route instance is created lazily for each connection. Calling the bare route name resolves its exposed init method; init is not called automatically before other methods.

add_event_listener(func: EventListenerType)¶

remove_event_listeners(func)¶

classmethod remove_route(route: str, fail=True)¶: Removes route by name. If fail=True an exception will be raised in case the route was not found.

property proxy¶

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 configure(keepalive_timeout=30, client_timeout=10, max_concurrent_requests=25, loads=json.loads, dumps=json.dumps)¶

classmethod freeze()¶

async authorize()¶: Return False to reject the WebSocket handshake with HTTP 403.

async classmethod serve(socket: WebSocketLike, *, context: Any = None)¶: Accept and serve a Starlette/FastAPI WebSocket until disconnect.

classmethod as_asgi(*, path: str | None = None)¶: Create an isolated ASGI endpoint with an optional path.

async run()¶

classmethod broadcast(func, callback=None, return_exceptions=True, **kwargs)¶

Route¶

class wsrpc_asgi.Route(socket: SocketT)¶

property socket¶

property loop¶

AllowedRoute¶

class wsrpc_asgi.AllowedRoute(socket: SocketT)¶

property socket¶

property loop¶

PrefixRoute¶

class wsrpc_asgi.PrefixRoute(socket: SocketT)¶

property socket¶

property loop¶

WebSocketRoute¶

class wsrpc_asgi.WebSocketRoute(socket: SocketT)¶

property socket¶

property loop¶

classmethod noproxy(func)¶

ClientException¶

class wsrpc_asgi.ClientException(payload)¶

WSRPCError¶

class wsrpc_asgi.WSRPCError¶

RPCApplication¶

class wsrpc_asgi.RPCApplication(handler: type[WebSocketBase], *, path: str | None = None)¶

Bare ASGI endpoint with its own registry and lifespan cleanup.

Available through Handler.as_asgi(). Keep the instance to broadcast only to its clients, or to close them explicitly in a parent lifespan.

get_clients()¶

broadcast(func, **kwargs)¶

async close()¶

RPCMiddleware¶

class wsrpc_asgi.RPCMiddleware(app, rpc: RPCApplication, *, path: str = '/ws/')¶

Route one WebSocket path to RPC and forward other scopes unchanged.

Install outside a framework to add RPC to any asyncio ASGI application. Framework middleware inside this wrapper does not run for the RPC path. Put shared authentication outside it or implement authorize.

Serialization¶

wsrpc_asgi.STATIC_DIR¶: Installed JavaScript clients and TypeScript declarations.

wsrpc_asgi.serializer(value)¶

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

from wsrpc_asgi import serializer

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

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

wsrpc_asgi.websocket.decorators.proxy(func: Callable[P, T])¶

wsrpc_asgi.websocket.decorators.noproxy(func: Callable[P, T])¶

wsrpc_asgi.signal.receiver(s: Signal[P])¶

class wsrpc_asgi.signal.Signal¶

connect(receiver: Callable[P, Coroutine[Any, Any, Any]])¶

call(*args: P.args, **kwargs: P.kwargs)¶

copy()¶

property is_frozen¶

freeze()¶