python client

This document describes how to use the python client library to communicate with a Volt via websocket.

Install packages

TODO - ask Ash how to package/publish this…

The repository is available on GitHub.

You will need to add the following dependencies to your poetry.toml file or equivalent:

... TBD ...

Usage

In order to be able to initialise a connection to a Volt, you will need to obtain a client configuration.

How your application obtains or stores the client configuration is not discussed here, needless to say you need to be careful not to expose keys or other sensitive data. For the purposes of this document, we assume the configuration is stored in a local file.

Once initialised successfully, you can issue any API call.

Refer to the API documentation for details of the method names and corresponding request and response messages. Each API method expects a JSON object representing the request as input, and returns a JSON object as indicated by the response message type. The JSON representation of the protobuf message format is intuitive.

Unary calls

All unary calls return a future that will either resolve with a JSON response object matching that defined by the API protobuf, or reject if there is a problem with the grpc transport or an error status on the response object.

An example of issuing a unary call is shown below:

import asyncio
import json
from volt_client_py.volt_client import VoltClient

async def main():
    # Load the configuration JSON from a file in the current directory.
    with open("volt.config.json", "r") as file:
        config = json.load(file)

    try:
        # Create a Volt client and initialise it with the configuration.
        client = VoltClient()
        await client.initialise(config)

        # Start an asyncio loop to enable the client to send and receive messages.
        loop = asyncio.gather(
            client.send_messages(),
            client.receive_messages(),
        )

        # Demonstrate fetching a resource (see https://docs.tdxvolt.com/en/api/volt_api#GetResource).
        homeFolderId = client.config["credential"]["identity_did"]
        resourceResponse = await client.GetResource({"resource_id": homeFolderId})
        print(f"GetResource() response: {resourceResponse}")

        #
        # Do other stuff...
        #

        # Wait for the main loop to finish
        await loop

        await client.close()

    except asyncio.CancelledError:
        print("async cancelled, waiting for next ping interval")

    except Exception as e:
        print(f"Failure: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Streaming calls

The promise/future model does not fit well with streaming calls since they are long-lived.

For this reason, all streaming calls (client streaming, server streaming, and bi-directional streaming) return a call object that is used to manage the call.

The call object emits events as interactions with the underlying websocket occur, and there are 3 events of interest:

“error” emitted when an error occurs on the call
“data” emitted when a response is received from the Volt
“end” emitted when the call ends

An example of a streaming call is shown below:

import asyncio
import traceback
import json
from volt_client_py.volt_client import VoltClient

async def main():
    # Load the configuration JSON from a file in the current directory.
    with open("volt.config.json", "r") as file:
        config = json.load(file)

    try:
        # Create a Volt client and initialise it
        client = VoltClient()
        await client.initialise(config)

        # Start an asyncio loop to enable the client to send and receive messages.
        loop = asyncio.gather(
            client.send_messages(),
            client.receive_messages(),
        )

        # Demonstrate wire subscription.
        wireSub = await client.SubscribeWire({"wire_id": "@wire"})
        print(f"SubscribeWire() response: {wireSub}")

        # Define a callback for the wire 'data' event.
        async def on_wire_data(data):
            print(f"Received wire data: {data}")
            await wireSub.end()

        # Define a callback for the wire 'end' event.
        def on_wire_end():
            print("Wire subscription ended")

        # Register the callbacks with the wire subscription.
        wireSub.on('data', on_wire_data)
        wireSub.on('end', on_wire_end)

        #
        # Do other stuff...
        #

        # Wait for the main loop to finish
        await loop

        await client.close()

    except asyncio.CancelledError:
        print("async cancelled, waiting for next ping interval")

    except Exception as e:
        print(f"Failure: {e}")
        traceback.print_exc()

if __name__ == "__main__":
    asyncio.run(main())

Connection

The library supports creating a bi-directional Connect call to the Volt. This can be used to register to receive events from the Volt and to implement services that the Volt or other clients can call.

An example of creating a connection is shown below:

import asyncio
import traceback
import json
from volt_client_py.volt_client import VoltClient
from volt_client_py.connection_manager import VoltConnectionManager

async def main():
    # Load the configuration JSON from a file in the current directory.
    with open("volt.config.json", "r") as file:
        config = json.load(file)

    try:
        # Create a Volt client and initialise it
        client = VoltClient()
        await client.initialise(config)

        # Create a connection manager and connect to the Volt. This is optional,
        # you only need to establish a connection if you want to receive notification
        # of Volt events or if you want to host a service.
        manager = VoltConnectionManager()
        await manager.connect(client)

        # Start an asyncio loop to enable the client to send and receive messages.
        loop = asyncio.gather(
            client.send_messages(),
            client.receive_messages(),
            manager.ping_connection()
        )

        # Define a callback for the 'connected' event.
        def on_connected(connected):
            print("Connected: ", connected)

        # Define a callback for 'event' events.
        def on_evt(evt):
            print(f"Received evt: {evt}")

        # Register the callbacks with the connection manager.
        manager.on('connected', on_connected)
        manager.on('event', on_evt)

        #
        # Do other stuff...
        #

        # Wait for the main loop to finish
        await loop

        await manager.disconnect()

        await client.close()

    except asyncio.CancelledError:
        print("async cancelled, waiting for next ping interval")

    except Exception as e:
        print(f"Failure: {e}")
        traceback.print_exc()

if __name__ == "__main__":
    asyncio.run(main())