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. Roadmap
6. Contributing
7. 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.