pynsq Documentation

pynsq is the official Python client library for NSQ. It offers two levels of abstraction for interacting with the NSQ protocol:

1. High-level API: Use the nsq.Reader and nsq.Writer classes to build robust consumers and producers with minimal boilerplate.
2. Low-level API: Use the async_conn module for synchronous or asynchronous communication directly over the NSQ TCP protocol. This is intended for developers who need to implement custom high-level functionality.

Note on Dependencies: If you use the async_conn module, you must have tornado installed, as the async module is built on top of the Tornado IOLoop.

Install the official Python client library for NSQ using pip.

$ pip install pynsq

The nsq.Reader class is the high-level consumer interface for interacting with NSQ. It provides an abstraction for connecting to an NSQ network, subscribing to topics, and consuming messages from channels. Use this class to manage the lifecycle of a consumer, including starting and stopping message processing.

import nsq

# Example usage of the Reader class
reader = nsq.Reader(topic='my_topic', channel='my_channel', handlers=[my_handler])
reader.connect()
# ... process messages ...
reader.stop()

The nsq.run() function is a convenience utility designed to simplify the execution of an NSQ consumer. It handles the setup and lifecycle management of the reader, making it suitable for simple scripts or entry points where you want to start consuming messages and keep the process running until interrupted.

import nsq

def my_handler(message):
    print(f'Received message: {message.body}')
    message.finish()

# Use run() to start the consumer loop
nsq.run(topic='my_topic', channel='my_channel', handlers=[my_handler])

The Client class is the primary entrypoint for managing connections to NSQ servers. It runs on the current Tornado IOLoop and includes a background periodic callback that checks for stale connections every 60 seconds. A connection is considered stale if it hasn't received data for more than twice its configured heartbeat_interval.

from nsq.client import Client

# Initialize the client
# Note: Client requires a running Tornado IOLoop
client = Client(name='my_client')

The heartbeat(conn) method is a hook provided by the Client class. It is called automatically whenever a heartbeat is received from an NSQ server. You can subclass Client and override this method to perform custom actions, such as updating monitoring metrics or logging liveness status.

:param conn: The nsq.AsyncConn instance over which the heartbeat was received.

from nsq.client import Client

class MyMonitoringClient(Client):
    def heartbeat(self, conn):
        # Perform custom action based on liveness
        print(f"Heartbeat received on connection: {conn.id}")
        # Call super() if you want to maintain default behavior
        super().heartbeat(conn)

client = MyMonitoringClient(name='monitor_client')