Requests API Reference¶

Request and Response classes for GNTP protocol communication.

Request Classes¶

BaseRequest¶

class BaseRequest(custom_headers=None, app_specific_headers=None)[source]¶

Bases: object

Base class for GNTP requests.

All GNTP requests inherit from this class and must define a message_type.

message_type¶

GNTP message type (REGISTER, NOTIFY, SUBSCRIBE)

Type:: str | None

custom_headers¶: List of (key, value) tuples for custom headers

app_specific_headers¶: List of (key, value) tuples for app headers

message_type: str | None = None¶

__init__(custom_headers=None, app_specific_headers=None)[source]¶

Initialize base request.

Parameters:

custom_headers (List[Tuple[str, Any]] | None) – Custom X- headers as (key, value) tuples
app_specific_headers (List[Tuple[str, Any]] | None) – App Data- headers as (key, value) tuples

write_into(writer)[source]¶

Serialize request into writer.

Subclasses must call this first, then add their specific fields.

Parameters:: writer (Any) – Message writer instance

__repr__()[source]¶

Return string representation.

RegisterRequest¶

class RegisterRequest(app_name, app_icon, events, custom_headers=None, app_specific_headers=None)[source]¶

Bases: BaseRequest

This request must be sent before notifications can be published. It registers the application and defines available notification types.

app_name¶: Application name

app_icon¶: Optional application icon

events¶: List of Event definitions

Example

from gntplib.requests import RegisterRequest
from gntplib import Event, Resource

events = [Event('update', 'Update Available')]
icon = Resource.from_file('app.png')
request = RegisterRequest('MyApp', icon, events)

message_type: str | None = 'REGISTER'¶

__init__(app_name, app_icon, events, custom_headers=None, app_specific_headers=None)[source]¶

Initialize REGISTER request.

Parameters:

app_name (str) – Name of the application
app_icon (Resource | None) – Optional icon resource for the application
events (List[Event]) – List of notification event definitions
custom_headers (List[Tuple[str, Any]] | None) – Custom X- headers
app_specific_headers (List[Tuple[str, Any]] | None) – App Data- headers

Example

>>> from gntplib import Event, Resource
>>> events = [Event('update', 'Update Available')]
>>> icon = Resource.from_file('app_icon.png')
>>> request = RegisterRequest('MyApp', icon, events)

write_into(writer)[source]¶

Serialize REGISTER request.

__repr__()[source]¶

Return string representation.

NotifyRequest¶

class NotifyRequest(app_name, notification, custom_headers=None, app_specific_headers=None)[source]¶

Bases: BaseRequest

NOTIFY request for sending a notification.

Sends an individual notification to the GNTP server. The application must be registered first.

app_name¶: Application name (must match registered name)

notification¶: Notification instance to send

Example

from gntplib.requests import NotifyRequest
from gntplib.models import Notification

notification = Notification('update', 'New Version', 'v2.0 released')
request = NotifyRequest('MyApp', notification)

message_type: str | None = 'NOTIFY'¶

__init__(app_name, notification, custom_headers=None, app_specific_headers=None)[source]¶

Initialize NOTIFY request.

Parameters:

app_name (str) – Name of the registered application
notification (Notification) – Notification to send
custom_headers (List[Tuple[str, Any]] | None) – Custom X- headers
app_specific_headers (List[Tuple[str, Any]] | None) – App Data- headers

Example

>>> from gntplib import Notification
>>> notif = Notification('update', 'New Version', 'v2.0 released')
>>> request = NotifyRequest('MyApp', notif)

write_into(writer)[source]¶

Serialize NOTIFY request.

__repr__()[source]¶

Return string representation.

SubscribeRequest¶

class SubscribeRequest(id_, name, port, custom_headers=None, app_specific_headers=None)[source]¶

Bases: BaseRequest

SUBSCRIBE request for subscribing to notifications.

Allows a client to subscribe to receive notifications from a hub.

id_¶: Unique subscriber identifier

name¶: Subscriber name

port¶: Port number for receiving notifications

message_type: str | None = 'SUBSCRIBE'¶

__init__(id_, name, port, custom_headers=None, app_specific_headers=None)[source]¶

Initialize SUBSCRIBE request.

Parameters:

id – Unique subscriber ID
name (str) – Subscriber name
port (int) – Port for receiving notifications
custom_headers (List[Tuple[str, Any]] | None) – Custom X- headers
app_specific_headers (List[Tuple[str, Any]] | None) – App Data- headers

Example

>>> request = SubscribeRequest('sub-123', 'MySubscriber', 23053)

write_into(writer)[source]¶

Serialize SUBSCRIBE request.

__repr__()[source]¶

Return string representation.

Response Class¶

class Response(message_type, headers)[source]¶

Bases: object

GNTP response from server.

Represents the response received from a GNTP server after sending a request.

message_type¶: Response type (-OK, -ERROR, -CALLBACK)

headers¶: Dictionary of response headers

Methods

is_ok()[source]

Check if response is successful.

Returns:: True if response type is -OK
Return type:: bool

is_error()[source]

Check if response is an error.

Returns:: True if response type is -ERROR
Return type:: bool

is_callback()[source]

Check if response is a callback.

Returns:: True if response type is -CALLBACK
Return type:: bool

get_header(name, default=None)[source]

Get header value with optional default.

Parameters:

name (str) – Header name
default (Any | None) – Default value if header not found

Returns:

Header value or default

Return type:

Any

__init__(message_type, headers)[source]¶

Initialize response.

Parameters:

message_type (str) – Response type (-OK, -ERROR, -CALLBACK)
headers (Dict[str, str]) – Response headers as dictionary

Raises:

GNTPResponseError – If response indicates an error

Example

>>> response = Response('-OK', {'Response-Action': 'REGISTER'})

is_ok()[source]¶

Check if response is successful.

Returns:: True if response type is -OK
Return type:: bool

is_error()[source]¶

Check if response is an error.

Returns:: True if response type is -ERROR
Return type:: bool

is_callback()[source]¶

Check if response is a callback.

Returns:: True if response type is -CALLBACK
Return type:: bool

get_header(name, default=None)[source]¶

Get header value with optional default.

Parameters:

name (str) – Header name
default (Any | None) – Default value if header not found

Returns:

Header value or default

Return type:

Any

__repr__()[source]¶

Return string representation.

__str__()[source]¶

Return human-readable string.

Utility Functions¶

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:

GNTPProtocolError – If response format is invalid
GNTPResponseError – If response indicates an error

Return type:

Response

Example

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