We first need an auth and a communication library in our virtualenv:

$pip install Authlib requests asyncio websockets


The example below highlights how to subscribe to location updates through the WebSocket API with the shared demo environment running the latest beta version of the DeepHub®:


from authlib.integrations.requests_client import OAuth2Session
import asyncio
import websockets
import json

# auth service variables
token_endpoint = 'https://dev.deephub.io/auth/realms/omlox/protocol/openid-connect/token'

# deephub service variables
websocket_endpoint = 'wss://dev.deephub.io/deephub/v1/ws/socket'

# auth client application variables
client_id = 'deephub-xxxx-client'
client_secret = 'xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxx'

async def listen_for_location_updates() -> None:

    # get a token from auth service
    openid_client = OAuth2Session(client_id, client_secret, scope='openid')
    token = openid_client.fetch_token(url=token_endpoint, client_id=client_id, client_secret=client_secret)

    # prove that we got an access_token actually
    if "access_token" in token:
        # form a websocket subscription payload to be sent
        payload = {
            "event": "subscribe",
            "topic": "location_updates",
            "params": {
                "token": token["access_token"]
            }
        }

        # open a websocket connection
        connection = websockets.connect(uri=websocket_endpoint)
        async with connection as websocket:
            # send subscription as string
            await websocket.send(json.dumps(payload))
            # wait for responses from websocket
            async for message in websocket:
                print(message)
            # close the connection
            await websocket.close()
    else:  # if access_token
        print("Authentication failed.")

# main --------------------------------
if __name__ == '__main__':
    asyncio.run(listen_for_location_updates())

By default, an access_token is valid for 15 minutes. After that, you should use the refresh_token to fetch new, fresh access_tokens (both are packed in the token above). In this example, we open a new session with every script start, which is not so optimal and should be handled in the PROD version on the OAuth2Session.