SignalR core client

Python signalr core client library, made by a guy born in Vilalba (Lugo).

About V1.0.0 (aka poluca)

Feature list:

• All kind of communications with the server (streaming, sending messages)
• All transports implemented (sse, long polling and web sockets)
• All encodings (text, binary - msgpack)
• Authentication
• Automatic reconnection with different strategies
• Custom ssl context passthrough (see certificates article)
• AsyncIO minimal implementation (will be improved on following versions)
• ...

Upcoming changes

• AsyncIO transport layer and callbacks
• Test suite, divide test into integration and unit. Making stubs of clients which enable testing without server
  • Managed solution azure server. For testing purposes only (PRs targeting main branch)
• Ack/Sequence implementation
• ...

Links

• Dev to posts with library examples and implementation

• Pypi

• Documentation

• Aspnetcore SignalR docs

Development

Software requirements:

• python >= 3.9
• virtualenv
• pip
• docker
• docker compose

Test environment has as a requirement a signalr core server, is available in here

Clone repos and install virtual environment:

git clone https://github.com/mandrewcito/signalrcore-containertestservers
cd signalrcore
make dev-install
git clone https://github.com/mandrewcito/signalrcore-containertestservers
cd signalrcore-containertestservers
docker compose up
cd ../signalrcore
make pytest-cov

Have fun :)

A Tiny How To

You can reach a lot of examples in tests folder, raw implementations in playground and fully working examples at the examples folder.

Connect to a server without auth

hub_connection = HubConnectionBuilder()\
    .with_url(server_url)\
    .configure_logging(logging.DEBUG)\
    .with_automatic_reconnect({
        "type": "raw",
        "keep_alive_interval": 10,
        "reconnect_interval": 5,
        "max_attempts": 5
    }).build()

Connect to a server with auth

login_function must provide auth token

hub_connection = HubConnectionBuilder()\
            .with_url(server_url,
            options={
                "access_token_factory": login_function,
                "headers": {
                    "mycustomheader": "mycustomheadervalue"
                }
            })\
            .configure_logging(logging.DEBUG)\
            .with_automatic_reconnect({
                "type": "raw",
                "keep_alive_interval": 10,
                "reconnect_interval": 5,
                "max_attempts": 5
            }).build()

Unauthorized errors

A login function must provide an error controller if authorization fails. When connection starts, if authorization fails exception will be propagated.

    def login(self):
        response = requests.post(
            self.login_url,
            json={
                "username": self.email,
                "password": self.password
                },verify=False)
        if response.status_code == 200:
            return response.json()["token"]
        raise requests.exceptions.ConnectionError()

    hub_connection.start()   # this code will raise  requests.exceptions.ConnectionError() if auth fails

Configure logging

HubConnectionBuilder()\
    .with_url(server_url,
    .configure_logging(logging.DEBUG)
    ...

Configure socket trace

HubConnectionBuilder()\
    .with_url(server_url,
    .configure_logging(logging.DEBUG, socket_trace=True) 
    ... 

Configure your own handler

 import logging
handler = logging.StreamHandler()
handler.setLevel(logging.DEBUG)
hub_connection = HubConnectionBuilder()\
    .with_url(server_url, options={"verify_ssl": False}) \
    .configure_logging(logging.DEBUG, socket_trace=True, handler=handler)
    ...

Configuring reconnection

After reaching max_attempts an exception will be thrown and on_disconnect event will be fired.

hub_connection = HubConnectionBuilder()\
    .with_url(server_url)\
    ...
    .build()

Configuring additional headers

hub_connection = HubConnectionBuilder()\
            .with_url(server_url,
            options={
                "headers": {
                    "mycustomheader": "mycustomheadervalue"
                }
            })
            ...
            .build()

Configuring additional querystring parameters

server_url ="http.... /?myQueryStringParam=134&foo=bar"
connection = HubConnectionBuilder()\
            .with_url(server_url,
            options={
            })\
            .build()

Configuring skip negotiation

hub_connection = HubConnectionBuilder() \
        .with_url("ws://"+server_url, options={
            "verify_ssl": False,
            "skip_negotiation": False,
            "headers": {
            }
        }) \
        .configure_logging(logging.DEBUG, socket_trace=True, handler=handler) \
        .build()

Configuring ping(keep alive)

keep_alive_interval sets the seconds of ping message

hub_connection = HubConnectionBuilder()\
    .with_url(server_url)\
    .configure_logging(logging.DEBUG)\
    .with_automatic_reconnect({
        "type": "raw",
        "keep_alive_interval": 10,
        "reconnect_interval": 5,
        "max_attempts": 5
    }).build()

Configuring logging

hub_connection = HubConnectionBuilder()\
    .with_url(server_url)\
    .configure_logging(logging.DEBUG)\
    .with_automatic_reconnect({
        "type": "raw",
        "keep_alive_interval": 10,
        "reconnect_interval": 5,
        "max_attempts": 5
    }).build()

Configure messagepack

from signalrcore.protocol.messagepack_protocol import MessagePackHubProtocol

HubConnectionBuilder()\
            .with_url(self.server_url, options={"verify_ssl":False})\
                ... 
            .with_hub_protocol(MessagePackHubProtocol())\
                ...
            .build()

Configure custom ssl context

You can add a custom ssl context to all requests and sockets

MY_CA_FILE_PATH = "ca.crt"
context = ssl.create_default_context(
    cafile=MY_CA_FILE_PATH
)

options = {
    "ssl_context": context
}

builder = HubConnectionBuilder()\
    .with_url(self.server_url, options=options)\
    .configure_logging(
        logging.INFO,
        socket_trace=True)
 
connection = builder.build()

More info about certificates here

Websockets

Will be used as transport layer by default, you do not need to specify it.

HubConnectionBuilder()\
    .with_url(server_http_url, options={
        ...
        "transport": HttpTransportType.web_sockets
        })\
    .configure_logging(logging.ERROR)\
    .build()

Server sent events

HubConnectionBuilder()\
    .with_url(server_http_url, options={
        ...
        "transport": HttpTransportType.server_sent_events
        })\
    .configure_logging(logging.ERROR)\
    .build()

Long polling

HubConnectionBuilder()\
    .with_url(server_http_url, options={
        ...
        "transport": HttpTransportType.long_polling
        })\
    .configure_logging(logging.ERROR)\
    .build()

Events

On Connect / On Disconnect

on_open - fires when connection is opened and ready to send messages on_close - fires when connection is closed

hub_connection.on_open(lambda: print("connection opened and handshake received ready to send messages"))
hub_connection.on_close(lambda: print("connection closed"))

On Hub Error (Hub Exceptions ...)

hub_connection.on_error(lambda data: print(f"An exception was thrown closed{data.error}"))

Register an operation

ReceiveMessage - signalr method print - function that has as parameters args of signalr method

hub_connection.on("ReceiveMessage", print)

Sending messages

SendMessage - signalr method username, message - parameters of signalrmethod

    hub_connection.send("SendMessage", [username, message])

Sending messages with callback

SendMessage - signalr method username, message - parameters of signalrmethod

    send_callback_received = threading.Lock()
    send_callback_received.acquire()
    self.connection.send(
        "SendMessage", # Method
        [self.username, self.message], # Params
        lambda m: send_callback_received.release()) # Callback
    if not send_callback_received.acquire(timeout=1):
        raise ValueError("CALLBACK NOT RECEIVED")

Requesting streaming (Server to client)

hub_connection.stream(
            "Counter",
            [len(self.items), 500]).subscribe({
                "next": self.on_next,
                "complete": self.on_complete,
                "error": self.on_error
            })

Client side Streaming

from signalrcore.subject import  Subject

subject = Subject()

# Start Streaming
hub_connection.send("UploadStream", subject)

# Each iteration
subject.next(str(iteration))

# End streaming
subject.complete()

AIO

Create connection

from signalrcore.aio.aio_hub_connection_builder import AIOHubConnectionBuilder

builder = AIOHubConnectionBuilder()\
    .with_url(self.server_url, options=options)\
    .configure_logging(
        self.get_log_level(),
        socket_trace=self.is_debug())\
    .with_automatic_reconnect({
        "type": "raw",
        "keep_alive_interval": 10,
        "reconnect_interval": 5,
        "max_attempts": 5
    })

hub = builder.build()

await hub.start()

await connection.send("SendMessage", [username, message])

await connection.stop()

Full Examples

Examples will be available here It were developed using package from aspnet core - SignalRChat

Chat example

A mini example could be something like this:

import logging
import sys
from signalrcore.hub_connection_builder import HubConnectionBuilder


def input_with_default(input_text, default_value):
    value = input(input_text.format(default_value))
    return default_value if value is None or value.strip() == "" else value


server_url = input_with_default('Enter your server url(default: {0}): ', "wss://localhost:44376/chatHub")
username = input_with_default('Enter your username (default: {0}): ', "mandrewcito")
handler = logging.StreamHandler()
handler.setLevel(logging.DEBUG)
hub_connection = HubConnectionBuilder()\
    .with_url(server_url, options={"verify_ssl": False}) \
    .configure_logging(logging.DEBUG, socket_trace=True, handler=handler) \
    .with_automatic_reconnect({
            "type": "interval",
            "keep_alive_interval": 10,
            "intervals": [1, 3, 5, 6, 7, 87, 3]
        }).build()

hub_connection.on_open(lambda: print("connection opened and handshake received ready to send messages"))
hub_connection.on_close(lambda: print("connection closed"))

hub_connection.on("ReceiveMessage", print)
hub_connection.start()
message = None

# Do login

while message != "exit()":
    message = input(">> ")
    if message is not None and message != "" and message != "exit()":
        hub_connection.send("SendMessage", [username, message])

hub_connection.stop()

sys.exit(0)