Core API Reference

This page documents the core classes and functions of gntplib.

Quick Functions

These convenience functions provide the simplest way to use gntplib.

publish(app_name, event_name, title, text='')[source]

Quick publish: Register and send notification in one call.

This is a convenience function that registers an application with a single event type and immediately sends a notification.

Parameters:

  • app_name (str) – Name of the application

  • event_name (str) – Name of the notification event

  • title (str) – Notification title

  • text (str) – Notification message (default: ‘’)

Example

>>> from gntplib import publish
>>> publish('MyApp', 'Alert', 'Warning', 'System overheating')
subscribe(id_, name, hub, password, port=23053)[source]

Quick subscribe: Send subscription request and return TTL.

Parameters:

  • id – Unique subscriber ID

  • name (str) – Subscriber name

  • hub (str | Tuple[str, int]) – Hub address (hostname or (host, port) tuple)

  • password (str) – Hub password

  • port (int) – Subscriber port (default: 23053)

Returns:

Subscription TTL in seconds

Return type:

int

Example

>>> from gntplib import subscribe
>>> ttl = subscribe('sub-123', 'MySubscriber', 'hub.example.com', 'secret')

Publisher Class

The main class for sending notifications.

class Publisher(name, event_defs, icon=None, custom_headers=None, app_specific_headers=None, gntp_client_class=None, **kwargs)[source]

Bases: BaseApp

GNTP notification publisher.

Registers an application and sends notifications to GNTP servers.

name

Application name

icon

Optional application icon

events

List of notification event definitions

Methods

register(callback=None)[source]

Register this publisher with the GNTP server.

Must be called before publishing notifications.

Parameters:

callback (Callable[[Response], None] | None) – Optional callback called with server response

Example

>>> pub = Publisher('MyApp', [Event('test')])
>>> pub.register()
publish(name, title, text='', id_=None, sticky=False, priority=0, icon=None, coalescing_id=None, callback=None, gntp_callback=None, **socket_callback_options)[source]

Send a notification.

Parameters:

  • name (str) – Event name (must be registered)

  • title (str) – Notification title

  • text (str) – Notification message (default: ‘’)

  • id – Unique notification ID

  • sticky (bool) – Keep until dismissed (default: False)

  • priority (int) – Priority -2 to 2 (default: 0)

  • icon (str | Resource | None) – Optional icon (URL or Resource)

  • coalescing_id (str | None) – ID for grouping notifications

  • callback (Callable | None) – Completion callback

  • gntp_callback (str | SocketCallback | None) – URL or SocketCallback for notification events

  • **socket_callback_options – Options for SocketCallback

Example

>>> pub.publish(
...     'update',
...     'New Version',
...     'Version 2.0 is available',
...     priority=1,
...     sticky=True
... )

Example

from gntplib import Publisher, Event

events = [Event('update', 'Update Available')]
publisher = Publisher('MyApp', events)
publisher.register()
publisher.publish('update', 'New Version', 'v2.0 released')
__init__(name, event_defs, icon=None, custom_headers=None, app_specific_headers=None, gntp_client_class=None, **kwargs)[source]

Initialize publisher.

Parameters:

  • name (str) – Application name

  • event_defs (List[str | Tuple[str, bool] | Event]) – Event definitions (str, tuple, or Event instances)

  • icon (str | Resource | None) – Optional app icon (URL or Resource)

  • custom_headers (List[Tuple[str, Any]] | None) – Custom X- headers

  • app_specific_headers (List[Tuple[str, Any]] | None) – App Data- headers

  • gntp_client_class (type | None) – Client class (default: GNTPClient)

  • **kwargs – Additional client arguments

Raises:

GNTPValidationError – If no events defined

Example

>>> from gntplib import Publisher, Event, Resource
>>> events = [Event('update', 'Update Available')]
>>> icon = Resource.from_file('icon.png')
>>> pub = Publisher('MyApp', events, icon=icon)
register(callback=None)[source]

Register this publisher with the GNTP server.

Must be called before publishing notifications.

Parameters:

callback (Callable[[Response], None] | None) – Optional callback called with server response

Example

>>> pub = Publisher('MyApp', [Event('test')])
>>> pub.register()
publish(name, title, text='', id_=None, sticky=False, priority=0, icon=None, coalescing_id=None, callback=None, gntp_callback=None, **socket_callback_options)[source]

Send a notification.

Parameters:

  • name (str) – Event name (must be registered)

  • title (str) – Notification title

  • text (str) – Notification message (default: ‘’)

  • id – Unique notification ID

  • sticky (bool) – Keep until dismissed (default: False)

  • priority (int) – Priority -2 to 2 (default: 0)

  • icon (str | Resource | None) – Optional icon (URL or Resource)

  • coalescing_id (str | None) – ID for grouping notifications

  • callback (Callable | None) – Completion callback

  • gntp_callback (str | SocketCallback | None) – URL or SocketCallback for notification events

  • **socket_callback_options – Options for SocketCallback

Example

>>> pub.publish(
...     'update',
...     'New Version',
...     'Version 2.0 is available',
...     priority=1,
...     sticky=True
... )
__repr__()[source]

Return string representation.

Subscriber Class

For subscribing to notifications from a GNTP hub.

class Subscriber(id_, name, hub, password, port=23053, custom_headers=None, app_specific_headers=None, gntp_client_class=None, **kwargs)[source]

Bases: BaseApp

GNTP notification subscriber.

Subscribes to receive notifications from a GNTP hub.

id_

Unique subscriber ID

name

Subscriber name

hub

Hub (host, port) tuple

password

Hub password

port

Subscriber port

ttl

Subscription time-to-live

Methods

subscribe(callback=None)[source]

Send subscription request to hub.

Updates ttl attribute with value from server response.

Parameters:

callback (Callable[[Response], None] | None) – Optional callback (default: updates ttl)

Example

>>> sub.subscribe()
>>> print(f"Subscription TTL: {sub.ttl} seconds")
store_ttl(response)[source]

Store TTL from subscription response.

Parameters:

response (Response) – Response from SUBSCRIBE request

Example

from gntplib import Subscriber

subscriber = Subscriber(
    id_='unique-id',
    name='MySubscriber',
    hub='hub.example.com',
    password='secret'
)
subscriber.subscribe()
print(f"TTL: {subscriber.ttl} seconds")
__init__(id_, name, hub, password, port=23053, custom_headers=None, app_specific_headers=None, gntp_client_class=None, **kwargs)[source]

Initialize subscriber.

Parameters:

  • id – Unique subscriber ID

  • name (str) – Subscriber name

  • hub (str | Tuple[str, int]) – Hub address (hostname or (host, port) tuple)

  • password (str) – Hub password

  • port (int) – Subscriber port (default: 23053)

  • custom_headers (List[Tuple[str, Any]] | None) – Custom X- headers

  • app_specific_headers (List[Tuple[str, Any]] | None) – App Data- headers

  • gntp_client_class (type | None) – Client class (default: GNTPClient)

  • **kwargs – Additional client arguments

Example

>>> sub = Subscriber(
...     'sub-123',
...     'MySubscriber',
...     'hub.example.com',
...     'secret'
... )
subscribe(callback=None)[source]

Send subscription request to hub.

Updates ttl attribute with value from server response.

Parameters:

callback (Callable[[Response], None] | None) – Optional callback (default: updates ttl)

Example

>>> sub.subscribe()
>>> print(f"Subscription TTL: {sub.ttl} seconds")
store_ttl(response)[source]

Store TTL from subscription response.

Parameters:

response (Response) – Response from SUBSCRIBE request

__repr__()[source]

Return string representation.

Authentication & Encryption

Keys Module

Authentication and key management for GNTP messages.

This module provides secure key generation and hashing for GNTP authentication. Supports MD5, SHA1, SHA256, and SHA512 hashing algorithms.

Hash Algorithms

MD5 = Algorithm('MD5', 16)

Factory class for creating authentication keys with specific hash algorithms.

This class represents a hashing algorithm configuration that can be used to create Key instances.

algorithm_id

String identifier for the algorithm (e.g., ‘SHA256’)

key_size

Size of the hash output in bytes

SHA1 = Algorithm('SHA1', 20)

Factory class for creating authentication keys with specific hash algorithms.

This class represents a hashing algorithm configuration that can be used to create Key instances.

algorithm_id

String identifier for the algorithm (e.g., ‘SHA256’)

key_size

Size of the hash output in bytes

SHA256 = Algorithm('SHA256', 32)

Factory class for creating authentication keys with specific hash algorithms.

This class represents a hashing algorithm configuration that can be used to create Key instances.

algorithm_id

String identifier for the algorithm (e.g., ‘SHA256’)

key_size

Size of the hash output in bytes

SHA512 = Algorithm('SHA512', 64)

Factory class for creating authentication keys with specific hash algorithms.

This class represents a hashing algorithm configuration that can be used to create Key instances.

algorithm_id

String identifier for the algorithm (e.g., ‘SHA256’)

key_size

Size of the hash output in bytes

class Key(password, algorithm_id='SHA256', salt=None)[source]

Bases: object

Authentication key for GNTP messages.

This class handles password-based authentication by generating secure hashes using a salt. The key is used for both authentication and optional encryption of GNTP messages.

password

Original password (stored as UTF-8 bytes)

algorithm_id

Hash algorithm identifier

salt

Random salt used in key derivation

key

Derived key from password and salt

key_hash

Hash of the derived key

__init__(password, algorithm_id='SHA256', salt=None)[source]

Initialize authentication key.

Parameters:

  • password (str) – Password for authentication

  • algorithm_id (str) – Hash algorithm (MD5, SHA1, SHA256, SHA512)

  • salt (bytes | None) – Optional salt. If None, random salt is generated

Raises:

ValueError – If algorithm_id is not supported

Example

>>> key = Key('mypassword', 'SHA256')
>>> key.key_hash_hex
b'...'
property salt_hex: bytes

Get hex-encoded salt.

Returns:

Hex-encoded salt as bytes

property key_hex: bytes

Get hex-encoded key.

Returns:

Hex-encoded key as bytes

property key_hash_hex: bytes

Get hex-encoded key hash.

Returns:

Hex-encoded key hash as bytes

verify_password(password)[source]

Verify if a password matches this key.

Parameters:

password (str) – Password to verify

Returns:

True if password matches, False otherwise

Return type:

bool

Example

>>> key = Key('secret', 'SHA256')
>>> key.verify_password('secret')
True
>>> key.verify_password('wrong')
False
__repr__()[source]

Return string representation of key.

__eq__(other)[source]

Check equality with another Key instance.

class Algorithm(algorithm_id, key_size)[source]

Bases: object

Factory class for creating authentication keys with specific hash algorithms.

This class represents a hashing algorithm configuration that can be used to create Key instances.

algorithm_id

String identifier for the algorithm (e.g., ‘SHA256’)

key_size

Size of the hash output in bytes

__init__(algorithm_id, key_size)[source]

Initialize algorithm configuration.

Parameters:

  • algorithm_id (str) – Algorithm identifier (MD5, SHA1, SHA256, SHA512)

  • key_size (int) – Expected hash output size in bytes

key(password, salt=None)[source]

Create a Key instance using this algorithm.

Parameters:

  • password (str) – Password for authentication

  • salt (bytes | None) – Optional salt bytes. If None, random salt is generated

Returns:

Key instance configured with this algorithm

Return type:

Key

Example

>>> algo = SHA256
>>> key = algo.key('mypassword')
__repr__()[source]

Return string representation of algorithm.

Key Class

class Key(password, algorithm_id='SHA256', salt=None)[source]

Bases: object

Authentication key for GNTP messages.

This class handles password-based authentication by generating secure hashes using a salt. The key is used for both authentication and optional encryption of GNTP messages.

password

Original password (stored as UTF-8 bytes)

algorithm_id

Hash algorithm identifier

salt

Random salt used in key derivation

key

Derived key from password and salt

key_hash

Hash of the derived key

Properties

property salt_hex: bytes

Get hex-encoded salt.

Returns:

Hex-encoded salt as bytes

property key_hex: bytes

Get hex-encoded key.

Returns:

Hex-encoded key as bytes

property key_hash_hex: bytes

Get hex-encoded key hash.

Returns:

Hex-encoded key hash as bytes

Methods

verify_password(password)[source]

Verify if a password matches this key.

Parameters:

password (str) – Password to verify

Returns:

True if password matches, False otherwise

Return type:

bool

Example

>>> key = Key('secret', 'SHA256')
>>> key.verify_password('secret')
True
>>> key.verify_password('wrong')
False

Example

from gntplib.keys import SHA256

key = SHA256.key('my-password')
print(f"Key hash: {key.key_hash_hex}")
__init__(password, algorithm_id='SHA256', salt=None)[source]

Initialize authentication key.

Parameters:

  • password (str) – Password for authentication

  • algorithm_id (str) – Hash algorithm (MD5, SHA1, SHA256, SHA512)

  • salt (bytes | None) – Optional salt. If None, random salt is generated

Raises:

ValueError – If algorithm_id is not supported

Example

>>> key = Key('mypassword', 'SHA256')
>>> key.key_hash_hex
b'...'
property salt_hex: bytes

Get hex-encoded salt.

Returns:

Hex-encoded salt as bytes

property key_hex: bytes

Get hex-encoded key.

Returns:

Hex-encoded key as bytes

property key_hash_hex: bytes

Get hex-encoded key hash.

Returns:

Hex-encoded key hash as bytes

verify_password(password)[source]

Verify if a password matches this key.

Parameters:

password (str) – Password to verify

Returns:

True if password matches, False otherwise

Return type:

bool

Example

>>> key = Key('secret', 'SHA256')
>>> key.verify_password('secret')
True
>>> key.verify_password('wrong')
False
__repr__()[source]

Return string representation of key.

__eq__(other)[source]

Check equality with another Key instance.

Ciphers Module

Encryption support for GNTP messages using PyCryptodome.

This module provides encryption capabilities for GNTP messages using AES, DES, and 3DES algorithms. Requires PyCryptodome package.

Note: This module requires PyCryptodome to be installed:

pip install pycryptodomex

Encryption Algorithms

AES = Algorithm('AES', 24)

Factory class for creating cipher instances.

algorithm_id

String identifier for the algorithm

key_size

Required key size in bytes

DES = Algorithm('DES', 8)

Factory class for creating cipher instances.

algorithm_id

String identifier for the algorithm

key_size

Required key size in bytes

DES3 = Algorithm('3DES', 24)

Factory class for creating cipher instances.

algorithm_id

String identifier for the algorithm

key_size

Required key size in bytes

class Cipher(key, algorithm_id='AES', iv=None)[source]

Bases: object

Message encryption and decryption using symmetric ciphers.

This class handles encryption of GNTP messages using AES, DES, or 3DES in CBC mode with PKCS7 padding.

key

Encryption key

algorithm_id

Algorithm identifier

algorithm

Cipher algorithm module

key_size

Required key size

iv

Initialization vector

__init__(key, algorithm_id='AES', iv=None)[source]

Initialize cipher with key and algorithm.

Parameters:

  • key (Key) – Key instance for encryption

  • algorithm_id (str) – Algorithm (AES, DES, 3DES)

  • iv (bytes | None) – Optional initialization vector. If None, random IV is generated

Raises:

GNTPEncryptionError – If algorithm is not supported or crypto unavailable

Example

>>> from gntplib import keys
>>> key = keys.SHA256.key('password')
>>> cipher = Cipher(key, 'AES')
property iv_hex: bytes

Get hex-encoded initialization vector.

Returns:

Hex-encoded IV as bytes

encrypt(plaintext)[source]

Encrypt plaintext data.

Parameters:

plaintext (bytes) – Data to encrypt

Returns:

Encrypted data

Raises:

GNTPEncryptionError – If encryption fails

Return type:

bytes

Example

>>> cipher = Cipher(key, 'AES')
>>> ciphertext = cipher.encrypt(b'secret message')
decrypt(ciphertext)[source]

Decrypt ciphertext data.

Parameters:

ciphertext (bytes) – Data to decrypt

Returns:

Decrypted plaintext

Raises:

GNTPEncryptionError – If decryption fails

Return type:

bytes

Example

>>> plaintext = cipher.decrypt(ciphertext)
__repr__()[source]

Return string representation.

__bool__()[source]

Check if cipher is active.

class Algorithm(algorithm_id, key_size)[source]

Bases: object

Factory class for creating cipher instances.

algorithm_id

String identifier for the algorithm

key_size

Required key size in bytes

__init__(algorithm_id, key_size)[source]

Initialize cipher algorithm configuration.

Parameters:

  • algorithm_id (str) – Algorithm identifier (AES, DES, 3DES)

  • key_size (int) – Required key size in bytes

cipher(key)[source]

Create a Cipher instance using this algorithm.

Parameters:

key (Key) – Key instance for encryption

Returns:

Cipher instance configured with this algorithm

Return type:

Cipher

Example

>>> from gntplib import keys, ciphers
>>> key = keys.SHA256.key('mypassword')
>>> cipher = ciphers.AES.cipher(key)
__repr__()[source]

Return string representation.

Cipher Class

class Cipher(key, algorithm_id='AES', iv=None)[source]

Bases: object

Message encryption and decryption using symmetric ciphers.

This class handles encryption of GNTP messages using AES, DES, or 3DES in CBC mode with PKCS7 padding.

key

Encryption key

algorithm_id

Algorithm identifier

algorithm

Cipher algorithm module

key_size

Required key size

iv

Initialization vector

Properties

property iv_hex: bytes

Get hex-encoded initialization vector.

Returns:

Hex-encoded IV as bytes

Methods

encrypt(plaintext)[source]

Encrypt plaintext data.

Parameters:

plaintext (bytes) – Data to encrypt

Returns:

Encrypted data

Raises:

GNTPEncryptionError – If encryption fails

Return type:

bytes

Example

>>> cipher = Cipher(key, 'AES')
>>> ciphertext = cipher.encrypt(b'secret message')
decrypt(ciphertext)[source]

Decrypt ciphertext data.

Parameters:

ciphertext (bytes) – Data to decrypt

Returns:

Decrypted plaintext

Raises:

GNTPEncryptionError – If decryption fails

Return type:

bytes

Example

>>> plaintext = cipher.decrypt(ciphertext)

Example

from gntplib.keys import SHA256
from gntplib.ciphers import AES

key = SHA256.key('password')
cipher = AES.cipher(key)

encrypted = cipher.encrypt(b'secret data')
decrypted = cipher.decrypt(encrypted)
__init__(key, algorithm_id='AES', iv=None)[source]

Initialize cipher with key and algorithm.

Parameters:

  • key (Key) – Key instance for encryption

  • algorithm_id (str) – Algorithm (AES, DES, 3DES)

  • iv (bytes | None) – Optional initialization vector. If None, random IV is generated

Raises:

GNTPEncryptionError – If algorithm is not supported or crypto unavailable

Example

>>> from gntplib import keys
>>> key = keys.SHA256.key('password')
>>> cipher = Cipher(key, 'AES')
property iv_hex: bytes

Get hex-encoded initialization vector.

Returns:

Hex-encoded IV as bytes

encrypt(plaintext)[source]

Encrypt plaintext data.

Parameters:

plaintext (bytes) – Data to encrypt

Returns:

Encrypted data

Raises:

GNTPEncryptionError – If encryption fails

Return type:

bytes

Example

>>> cipher = Cipher(key, 'AES')
>>> ciphertext = cipher.encrypt(b'secret message')
decrypt(ciphertext)[source]

Decrypt ciphertext data.

Parameters:

ciphertext (bytes) – Data to decrypt

Returns:

Decrypted plaintext

Raises:

GNTPEncryptionError – If decryption fails

Return type:

bytes

Example

>>> plaintext = cipher.decrypt(ciphertext)
__repr__()[source]

Return string representation.

__bool__()[source]

Check if cipher is active.

Client & Connection

GNTP Client

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

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

Methods

process_request(request, callback=None, **kwargs)[source]

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)

Example

from gntplib.connections import GNTPClient
from gntplib.requests import RegisterRequest
from gntplib import Event

client = GNTPClient(
    host='localhost',
    port=23053,
    password='secret',
    key_hashing=SHA256,
    encryption=AES
)

request = RegisterRequest('MyApp', None, [Event('test')])
client.process_request(request)
__init__(host='localhost', port=23053, timeout=10.0, password=None, key_hashing=None, encryption=None, connection_class=None)[source]

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)[source]

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__()[source]

Return string representation.

GNTP Connection

class GNTPConnection(address, timeout, final_callback=None, socket_callback=None)[source]

Bases: BaseGNTPConnection

Synchronous GNTP connection implementation.

Provides blocking socket communication with GNTP server.

sock

TCP socket connection

Methods

write_message(message)[source]

Send message to GNTP server.

Parameters:

message (bytes) – Message bytes to send

Raises:

GNTPConnectionError – If send fails

read_message(callback)[source]

Read message from server and call callback.

Parameters:

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

Raises:

GNTPConnectionError – If read fails

close()[source]

Close the socket connection.

__init__(address, timeout, final_callback=None, socket_callback=None)[source]

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)[source]

Send message to GNTP server.

Parameters:

message (bytes) – Message bytes to send

Raises:

GNTPConnectionError – If send fails

read_message(callback)[source]

Read message from server and call callback.

Parameters:

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

Raises:

GNTPConnectionError – If read fails

close()[source]

Close the socket connection.

Base Connection

class BaseGNTPConnection(final_callback=None, socket_callback=None)[source]

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

Callback Methods

on_ok_message(message)[source]

Handle -OK response from server.

Parameters:

message (bytes) – Response message bytes

on_callback_message(message)[source]

Handle -CALLBACK response from server.

Parameters:

message (bytes) – Callback message bytes

__init__(final_callback=None, socket_callback=None)[source]

Initialize base connection.

Parameters:

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

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

on_ok_message(message)[source]

Handle -OK response from server.

Parameters:

message (bytes) – Response message bytes

on_callback_message(message)[source]

Handle -CALLBACK response from server.

Parameters:

message (bytes) – Callback message bytes

write_message(message)[source]

Send message to server.

Subclasses must implement this method.

Parameters:

message (bytes) – Message bytes to send

read_message(callback)[source]

Read message from server.

Subclasses must implement this method.

Parameters:

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

close()[source]

Close the connection.

Subclasses must implement this method.

Utility Functions

Message Generation

generate_messages(sock, buffer_size=1024)[source]

Generate complete GNTP messages from socket.

Reads from socket and yields complete messages terminated by double CRLF (rnrn).

Parameters:

  • sock (socket) – Socket to read from

  • buffer_size (int) – Read buffer size in bytes

Yields:

Complete GNTP message bytes

Raises:

GNTPError – If message exceeds maximum size

Example

import socket
from gntplib.connections import generate_messages

sock = socket.create_connection(('localhost', 23053))
sock.send(request_message)

for message in generate_messages(sock):
    print(f"Received: {message}")
    break

Response Parsing

parse_response(message, expected_type=None)[source]

Parse GNTP response message.

Parameters:

  • message (bytes) – Raw response message bytes

  • expected_type (str | None) – Expected response type for validation

Returns:

Response instance

Raises:
Return type:

Response

Example

>>> response = parse_response(b'GNTP/1.0 -OK NONE\r\n...')

Example

from gntplib.requests import parse_response

message = b'GNTP/1.0 -OK NONE\r\nResponse-Action: REGISTER\r\n\r\n'
response = parse_response(message, expected_type='-OK')
print(response.headers)

Constants

Protocol Constants

Constants and utility functions for GNTP library.

This module defines all constants used in the GNTP protocol and provides utility functions for common operations.

Network Defaults

DEFAULT_PORT = 23053

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

DEFAULT_TIMEOUT = 10.0

Convert a string or number to a floating point number, if possible.

DEFAULT_TTL = 60

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

Message Limits

MAX_MESSAGE_SIZE = 4096

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

MAX_LINE_SIZE = 1024

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

Delimiters

LINE_DELIMITER = b'\r\n'

bytes(iterable_of_ints) -> bytes bytes(string, encoding[, errors]) -> bytes bytes(bytes_or_buffer) -> immutable copy of bytes_or_buffer bytes(int) -> bytes object of size given by the parameter initialized with null bytes bytes() -> empty bytes object

Construct an immutable array of bytes from:

  • an iterable yielding integers in range(256)

  • a text string encoded using the specified encoding

  • any object implementing the buffer API.

  • an integer

MESSAGE_DELIMITER = b'\r\n\r\n'

bytes(iterable_of_ints) -> bytes bytes(string, encoding[, errors]) -> bytes bytes(bytes_or_buffer) -> immutable copy of bytes_or_buffer bytes(int) -> bytes object of size given by the parameter initialized with null bytes bytes() -> empty bytes object

Construct an immutable array of bytes from:

  • an iterable yielding integers in range(256)

  • a text string encoded using the specified encoding

  • any object implementing the buffer API.

  • an integer

Priority Levels

PRIORITY_VERY_LOW = -2

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

PRIORITY_LOW = -1

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

PRIORITY_NORMAL = 0

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

PRIORITY_HIGH = 1

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

PRIORITY_EMERGENCY = 2

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

Response Types

RESPONSE_OK = '-OK'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

RESPONSE_ERROR = '-ERROR'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

RESPONSE_CALLBACK = '-CALLBACK'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

random_bytes(num_bytes)[source]

Generate cryptographically secure random bytes.

This function uses secrets module for secure random generation, suitable for cryptographic operations.

Parameters:

num_bytes (int) – Number of random bytes to generate

Returns:

Bytes object containing random data

Raises:

ValueError – If num_bytes is negative

Return type:

bytes

Example

>>> data = random_bytes(16)
>>> len(data)
16
random_hex_string(num_bytes)[source]

Generate a random hex string.

Parameters:

num_bytes (int) – Number of bytes to generate (hex string will be 2x this length)

Returns:

Hex string representation of random bytes

Return type:

str

Example

>>> hex_str = random_hex_string(8)
>>> len(hex_str)
16
encode_utf8(text)[source]

Safely encode text to UTF-8 bytes.

Parameters:

text (str) – String to encode

Returns:

UTF-8 encoded bytes

Return type:

bytes

Example

>>> encode_utf8('Hello')
b'Hello'
decode_utf8(data)[source]

Safely decode UTF-8 bytes to string.

Parameters:

data (bytes) – Bytes to decode

Returns:

Decoded string

Raises:

UnicodeDecodeError – If data is not valid UTF-8

Return type:

str

Example

>>> decode_utf8(b'Hello')
'Hello'
safe_int_conversion(value, default=0)[source]

Safely convert a value to integer with fallback.

Parameters:

  • value (any) – Value to convert to int

  • default (int) – Default value if conversion fails

Returns:

Integer value or default

Return type:

int

Example

>>> safe_int_conversion('42')
42
>>> safe_int_conversion('invalid', default=-1)
-1
validate_priority(priority)[source]

Validate and clamp priority value to valid range.

Priority must be between -2 and 2 (inclusive).

Parameters:

priority (int) – Priority value to validate

Returns:

Clamped priority value

Return type:

int

Example

>>> validate_priority(5)
2
>>> validate_priority(-5)
-2

Utility Functions

random_bytes(num_bytes)[source]

Generate cryptographically secure random bytes.

This function uses secrets module for secure random generation, suitable for cryptographic operations.

Parameters:

num_bytes (int) – Number of random bytes to generate

Returns:

Bytes object containing random data

Raises:

ValueError – If num_bytes is negative

Return type:

bytes

Example

>>> data = random_bytes(16)
>>> len(data)
16
random_hex_string(num_bytes)[source]

Generate a random hex string.

Parameters:

num_bytes (int) – Number of bytes to generate (hex string will be 2x this length)

Returns:

Hex string representation of random bytes

Return type:

str

Example

>>> hex_str = random_hex_string(8)
>>> len(hex_str)
16
encode_utf8(text)[source]

Safely encode text to UTF-8 bytes.

Parameters:

text (str) – String to encode

Returns:

UTF-8 encoded bytes

Return type:

bytes

Example

>>> encode_utf8('Hello')
b'Hello'
decode_utf8(data)[source]

Safely decode UTF-8 bytes to string.

Parameters:

data (bytes) – Bytes to decode

Returns:

Decoded string

Raises:

UnicodeDecodeError – If data is not valid UTF-8

Return type:

str

Example

>>> decode_utf8(b'Hello')
'Hello'
validate_priority(priority)[source]

Validate and clamp priority value to valid range.

Priority must be between -2 and 2 (inclusive).

Parameters:

priority (int) – Priority value to validate

Returns:

Clamped priority value

Return type:

int

Example

>>> validate_priority(5)
2
>>> validate_priority(-5)
-2

Example

from gntplib.constants import validate_priority

# Clamps to valid range [-2, 2]
priority = validate_priority(10)  # Returns 2
priority = validate_priority(-5)  # Returns -2

Helper Functions

coerce_to_events(items)[source]

Convert various event definitions to Event instances.

Parameters:

items (List[str | Tuple[str, bool] | Event]) – List of event definitions - str: Event name (enabled by default) - (str, bool): (Event name, enabled flag) - Event: Event instance (used as-is)

Returns:

List of Event instances

Return type:

List[Event]

Example

>>> events = coerce_to_events(['event1', ('event2', False)])

Example

from gntplib.lib import coerce_to_events
from gntplib import Event

# Mixed event definitions
events = coerce_to_events([
    'simple_event',              # String
    ('named_event', False),      # Tuple (name, enabled)
    Event('full_event', icon=icon)  # Event object
])
coerce_to_callback(gntp_callback=None, **socket_callback_options)[source]

Convert callback specification to callback instance.

Parameters:

  • gntp_callback (str | SocketCallback | None) – URL string or SocketCallback instance

  • **socket_callback_options – Options for creating SocketCallback

Returns:

URLCallback, SocketCallback, or None

Raises:

GNTPError – If both gntp_callback and options are provided

Return type:

URLCallback | SocketCallback | None

Example

from gntplib.lib import coerce_to_callback

# URL callback
callback = coerce_to_callback('https://example.com/notify')

# Socket callback
callback = coerce_to_callback(
    on_click=handle_click,
    on_close=handle_close
)

See Also