Connections API Reference

Connection and client classes for GNTP communication.

See Core API Reference for the main documentation of these classes.

GNTP connection and client implementations.

This module provides synchronous connection handling and client logic for communicating with GNTP servers.

class BaseGNTPConnection(final_callback=None, socket_callback=None)

Bases: object

Abstract base class for GNTP connections.

Defines the interface for GNTP server connections. Subclasses must implement write_message, read_message, and close methods.

final_callback

Callback when connection completes

socket_callback

Optional callback for socket events

__init__(final_callback=None, socket_callback=None)

Initialize base connection.

Parameters:

• final_callback (Callable | None) – Called when operation completes

• socket_callback (Any | None) – Optional socket event callback

on_ok_message(message)

Handle -OK response from server.

Parameters:

message (bytes) – Response message bytes

on_callback_message(message)

Handle -CALLBACK response from server.

Parameters:

message (bytes) – Callback message bytes

write_message(message)

Send message to server.

Subclasses must implement this method.

Parameters:

message (bytes) – Message bytes to send

read_message(callback)

Read message from server.

Subclasses must implement this method.

Parameters:

callback (Callable[[bytes], None]) – Function to call with received message

close()

Close the connection.

Subclasses must implement this method.

class GNTPConnection(address, timeout, final_callback=None, socket_callback=None)

Bases: BaseGNTPConnection

Synchronous GNTP connection implementation.

Provides blocking socket communication with GNTP server.

sock

TCP socket connection

__init__(address, timeout, final_callback=None, socket_callback=None)

Initialize GNTP connection.

Parameters:

• address (Tuple[str, int]) – (host, port) tuple

• timeout (float) – Connection timeout in seconds

• final_callback (Callable | None) – Called when operation completes

• socket_callback (Any | None) – Optional socket event callback

Raises:

GNTPConnectionError – If connection fails

write_message(message)

Send message to GNTP server.

Parameters:

message (bytes) – Message bytes to send

Raises:

GNTPConnectionError – If send fails

read_message(callback)

Read message from server and call callback.

Parameters:

callback (Callable[[bytes], None]) – Function to call with received message

Raises:

GNTPConnectionError – If read fails

close()

Close the socket connection.

class GNTPClient(host='localhost', port=23053, timeout=10.0, password=None, key_hashing=None, encryption=None, connection_class=None)

Bases: object

Synchronous GNTP client.

Handles request processing with optional authentication and encryption.

address

(host, port) of GNTP server

timeout

Connection timeout in seconds

connection_class

Connection class to use

packer_factory

Factory for creating message packers

__init__(host='localhost', port=23053, timeout=10.0, password=None, key_hashing=None, encryption=None, connection_class=None)

Initialize GNTP client.

Parameters:

• host (str) – GNTP server hostname

• port (int) – GNTP server port

• timeout (float) – Connection timeout in seconds

• password (str | None) – Optional password for authentication

• key_hashing (Algorithm | None) – Key hashing algorithm (default: SHA256)

• encryption (Algorithm | None) – Optional encryption algorithm

• connection_class (type | None) – Connection class (default: GNTPConnection)

Raises:

GNTPError – If encryption key size is too large for hashing

Example

>>> from gntplib import keys, ciphers
>>> client = GNTPClient(
...     host='localhost',
...     password='secret',
...     key_hashing=keys.SHA256,
...     encryption=ciphers.AES
... )

process_request(request, callback=None, **kwargs)

Process a GNTP request.

Serializes the request, sends it to the server, and handles the response.

Parameters:

• request (Any) – Request object to send

• callback (Callable | None) – Optional callback when operation completes

• **kwargs – Additional arguments passed to connection

Example

>>> from gntplib import RegisterRequest, Event
>>> request = RegisterRequest('MyApp', None, [Event('test')])
>>> client.process_request(request)

__repr__()

Return string representation.