PySignalR Client

Warnings

• breaking Python 3.9 support was officially dropped in `pysignalr` version 1.3.0. Users on Python 3.9 or older must upgrade their Python environment to at least 3.10 to use versions 1.3.0 and newer.
  Fix: Upgrade Python to version 3.10 or newer. If unable to upgrade, pin `pysignalr` to a version prior to 1.3.0 (e.g., `pysignalr<1.3.0`).
• gotcha Version 1.3.1 introduced several fixes for SignalR Hub Protocol spec compliance in JSON and MessagePack codecs (e.g., `streamIds`, `invocationId`, `ResultKind`, headers, varint framing). Older versions might have exhibited non-compliant behavior that could lead to subtle issues or unexpected interactions with some SignalR servers.
  Fix: Upgrade to `pysignalr` version 1.3.1 or newer to ensure full protocol compliance and stability.
• gotcha Prior to version 1.1.0, the reconnection logic in `pysignalr` was prone to issues. Applications relying on stable and automatic reconnections in the face of network interruptions might experience unreliability in older versions.
  Fix: Upgrade to `pysignalr` version 1.1.0 or newer to benefit from improved reconnection stability.
• gotcha SignalR itself can silently fail to invoke client methods if the method name or signature sent from the server does not exactly match a registered client-side handler. The server will not receive an error.
  Fix: Ensure exact matching of method names and parameter types/counts between server invocations and client `on()` handlers. Enable client-side logging (`logging.DEBUG`) to diagnose unhandled messages or errors.
• gotcha The `access_token_factory` argument was added in 1.1.0, allowing dynamic token generation for authentication. If you were using older, less flexible authentication methods or hardcoding tokens, you might need to refactor your authentication logic when upgrading to leverage this feature or if your token needs refreshing.
  Fix: Adopt the `access_token_factory` for dynamic token management, especially with expiring JWTs, to ensure continuous authentication without manual re-connection.

Install

• pip install pysignalr Install stable version

Imports

• SignalRClient

  from pysignalr.client import SignalRClient

• CompletionMessage

  from pysignalr.messages import CompletionMessage

Quickstart

This quickstart demonstrates how to establish a connection to a SignalR hub, register event handlers for incoming messages, and send messages to the server. It includes examples for connection lifecycle events (open, close, error) and handling specific server events, including those requesting a client result.

import asyncio
from contextlib import suppress
from typing import Any, Dict, List
from pysignalr.client import SignalRClient
from pysignalr.messages import CompletionMessage
import os

async def on_open() -> None:
    print('Connected to the server')

async def on_close() -> None:
    print('Disconnected from the server')

async def on_message(message: List[Dict[str, Any]]) -> None:
    print(f'Received message: {message}')

async def on_client_result(message: list[dict[str, Any]]) -> str:
    print(f'Received message requesting result: {message}')
    return 'reply_from_client'

async def on_error(message: CompletionMessage) -> None:
    print(f'Received error: {message.error}')

async def main() -> None:
    # Replace with your SignalR hub URL
    # For example, a public API like TzKT.io or your own local/remote hub
    signalr_url = os.environ.get('SIGNALR_HUB_URL', 'https://api.tzkt.io/v1/ws')
    access_token = os.environ.get('SIGNALR_ACCESS_TOKEN', '') # Optional: for authenticated hubs

    client_args = {'url': signalr_url}
    if access_token:
        # For authenticated hubs, provide an access token factory
        client_args['access_token_factory'] = lambda: access_token

    client = SignalRClient(**client_args)
    client.on_open(on_open)
    client.on_close(on_close)
    client.on_error(on_error)

    # Register handlers for specific events from the server
    client.on('operations', on_message) # Example: subscribing to 'operations' event
    client.on('client_result', on_client_result) # Example: handling a server request for a client result

    await asyncio.gather(
        client.run(),
        # Example: Sending a message to the server (e.g., to subscribe to a topic)
        client.send('SubscribeToOperations', [{}]), 
    )

if __name__ == '__main__':
    with suppress(KeyboardInterrupt, asyncio.CancelledError):
        asyncio.run(main())