pysignalr

pysignalr

pysignalr is a modern, reliable, and async-ready client for the SignalR protocol. This project started as an asyncio fork of mandrewcito's signalrcore library and ended up as a complete rewrite.

Table of Contents

1. Installation
2. Basic Usage
3. Usage with Token Authentication
4. API Reference
5. License

Installation

To install pysignalr, simply use pip:

pip install pysignalr

Basic Usage

Let's connect to TzKT, an API and block explorer of Tezos blockchain, and subscribe to all operations:

import asyncio
from contextlib import suppress
from typing import Any
from typing import Dict
from typing import List

from pysignalr.client import SignalRClient
from pysignalr.messages import CompletionMessage


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_error(message: CompletionMessage) -> None:
    print(f'Received error: {message.error}')


async def main() -> None:
    client = SignalRClient('https://api.tzkt.io/v1/ws')

    client.on_open(on_open)
    client.on_close(on_close)
    client.on_error(on_error)
    client.on('operations', on_message)

    await asyncio.gather(
        client.run(),
        client.send('SubscribeToOperations', [{}]),
    )


with suppress(KeyboardInterrupt, asyncio.CancelledError):
    asyncio.run(main())

Usage with Token Authentication

To connect to the SignalR server using token authentication:

import asyncio
from contextlib import suppress
from typing import Any, Dict, List

from pysignalr.client import SignalRClient
from pysignalr.messages import CompletionMessage

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_error(message: CompletionMessage) -> None:
    print(f'Received error: {message.error}')

def token_factory() -> str:
    # Replace with logic to fetch or generate the token
    return "your_access_token_here"

async def main() -> None:
    client = SignalRClient(
        url='https://api.tzkt.io/v1/ws',
        access_token_factory=token_factory,
        headers={"mycustomheader": "mycustomheadervalue"},
    )

    client.on_open(on_open)
    client.on_close(on_close)
    client.on_error(on_error)
    client.on('operations', on_message)

    await asyncio.gather(
        client.run(),
        client.send('SubscribeToOperations', [{}]),
    )

with suppress(KeyboardInterrupt, asyncio.CancelledError):
    asyncio.run(main())

API Reference

SignalRClient

Parameters

• url (str): The SignalR server URL.
• access_token_factory (Callable[[], str], optional): A function that returns the access token.
• headers (Dict[str, str], optional): Additional headers to include in the WebSocket handshake.

Methods

• on_open(callback: Callable[[], Awaitable[None]]): Set the callback for connection open event.
• on_close(callback: Callable[[], Awaitable[None]]): Set the callback for connection close event.
• on_error(callback: Callable[[CompletionMessage], Awaitable[None]]): Set the callback for error events.
• on(event: str, callback: Callable[[List[Dict[str, Any]]], Awaitable[None]]): Set the callback for a specific event.
• send(method: str, args: List[Any]): Send a message to the server.

CompletionMessage

A message received from the server upon completion of a method invocation.

Attributes

• error (str): The error message, if any.

License

This project is licensed under the MIT License - see the LICENSE file for details.