DOCSEARCHbyAlgolia
Submit Docs

fastapi_websocket_rpc

repository·master·Indexed 42 days ago

https://github.com/permitio/fastapi_websocket_rpc

A Python library providing a bidirectional JSON RPC channel over WebSockets for FastAPI servers and Python clients. It enables the exposure and calling of asynchronous methods on either side of the connection using the RpcMethodsBase class and the .other property for remote method invocation.

Tokens
1.5K
Snippets
7
Records
7
Agent score
4%

What's inside fastapi_websocket_rpc

  1. Call remote methods using the .other property

    master

    Once a connection is established, you can call methods defined on the remote side (server or client) using the .other wrapper. The call returns a response object where the return value of the remote method is available via the .result attribute.

    Note: Currently, only keyword arguments are supported for RPC method calls.

    # Example: Calling an 'add' method on the remote side
response = await client.other.add(a=1, b=2)
print(response.result) # 3

  2. Enable HTTP(S) Proxy support for the Client

    master

    The default websockets client handler does not support HTTP(S) proxies. To enable proxy support, install the extra dependency and use the ProxyEnabledWebSocketClientHandler class.

    pip install fastapi_websocket_rpc[websocket-client]
    import asyncio
from fastapi_websocket_rpc import RpcMethodsBase, WebSocketRpcClient, ProxyEnabledWebSocketClientHandler

async def run_client(uri):
    async with WebSocketRpcClient(uri, RpcMethodsBase(), websocket_client_handler_cls = ProxyEnabledWebSocketClientHandler) as client:
        # Use standard environment variables for proxy configuration:
        # http_proxy, https_proxy, and no_proxy
        pass

  3. How to implement a WebSocket RPC Client

    master

    To create a client, use the WebSocketRpcClient as an async context manager. Pass the server URI and a class inheriting from RpcMethodsBase (to define local methods if needed) to the constructor. Remote methods are accessed via the .other property of the client instance.

    import asyncio
from fastapi_websocket_rpc import RpcMethodsBase, WebSocketRpcClient

async def run_client(uri):
    async with WebSocketRpcClient(uri, RpcMethodsBase()) as client:
        # call concat on the other side
        response = await client.other.concat(a="hello", b=" world")
        # print result
        print(response.result)  # will print "hello world"

# run the client until it completes interaction with server
asyncio.get_event_loop().run_until_complete(
    run_client("ws://localhost:9000/ws")
)

  5. How to implement a WebSocket RPC Server

    master

    To create a server, follow these steps:

    1. Define a class inheriting from RpcMethodsBase containing the async methods you want to expose.
    2. Initialize a FastAPI application.
    3. Create a WebsocketRPCEndpoint instance, passing your methods class to it.
    4. Register the endpoint to your FastAPI app using endpoint.register_route(app, "/path").
    5. Run the server using uvicorn.
    import uvicorn
from fastapi import FastAPI
from fastapi_websocket_rpc import RpcMethodsBase, WebsocketRPCEndpoint

# Methods to expose to the clients
class ConcatServer(RpcMethodsBase):
    async def concat(self, a="", b=""):
        return a + b
    
# Init the FAST-API app
app =  FastAPI()
# Create an endpoint and load it with the methods to expose
endpoint = WebsocketRPCEndpoint(ConcatServer())
# add the endpoint to the app
endpoint.register_route(app, "/ws")

# Start the server itself
uvicorn.run(app, host="0.0.0.0", port=9000)

  6. Configure RPC logging

    master

    You can control the logging output of the library using the logging_config helper or the WS_RPC_LOGGING environment variable. You can set the mode to match other tools like UVICORN.

    # set RPC to log like UVICORN
from fastapi_websocket_rpc.logger import logging_config, LoggingModes
logging_config.set_mode(LoggingModes.UVICORN)

  7. Define RPC methods with RpcMethodsBase

    master

    To expose functionality to the other side of the connection, create a class that inherits from RpcMethodsBase. Any async methods defined in this class can be called remotely.

    Constraints:

    • Methods must be async.
    • Currently, only keyword arguments are supported for these methods.
    from fastapi_websocket_rpc import RpcMethodsBase

class MyMethods(RpcMethodsBase):
    async def my_remote_method(self, param1, param2):
        return f"Received {param1} and {param2}"