PubNub Python SDK

10.6.2 · active · verified Thu Apr 16

The PubNub Python SDK provides real-time messaging, presence, and push services. It enables developers to build interactive applications with features like chat, IoT, and live event streaming. The current stable version is 10.6.2, with active development and frequent releases addressing bug fixes and feature enhancements, typically on a monthly to quarterly cadence.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart initializes the PubNub client, subscribes to a channel, and publishes a message. It demonstrates setting up `PNConfiguration`, attaching a `SubscribeCallback` for handling messages, status, and presence events, and performing basic publish/subscribe operations. Remember to replace demo keys with your own.

import os
import time
from pubnub.pubnub import PubNub
from pubnub.pnconfiguration import PNConfiguration
from pubnub.callbacks import SubscribeCallback
from pubnub.enums import PNStatusCategory

# Replace with your actual publish and subscribe keys
publish_key = os.environ.get('PUBNUB_PUBLISH_KEY', 'demo')
subscribe_key = os.environ.get('PUBNUB_SUBSCRIBE_KEY', 'demo')

if publish_key == 'demo' or subscribe_key == 'demo':
    print("WARNING: Using demo keys. Replace with your own PubNub keys for production.")

pnconfig = PNConfiguration()
pnconfig.publish_key = publish_key
pnconfig.subscribe_key = subscribe_key
pnconfig.uuid = "my_unique_uuid_py"

pnub = PubNub(pnconfig)

class MySubscribeCallback(SubscribeCallback):
    def presence(self, pubnub, presence):
        print(f"PRESENCE: {presence.event}")

    def status(self, pubnub, status):
        if status.category == PNStatusCategory.PNUnexpectedDisconnectCategory:
            print("Disconnected.")
        elif status.category == PNStatusCategory.PNConnectedCategory:
            print("Connected to PubNub.")
        elif status.category == PNStatusCategory.PNReconnectedCategory:
            print("Reconnected to PubNub.")
        else:
            print(f"STATUS: {status.category}")

    def message(self, pubnub, message):
        print(f"MESSAGE: {message.message} from {message.channel}")

pnub.add_listener(MySubscribeCallback())
pnub.subscribe().channels('my_channel').with_presence().execute()

try:
    print("Publishing 'hello world' in 5 seconds...")
    time.sleep(5)
    pubnub.publish().channel('my_channel').message({'text': 'hello world', 'timestamp': time.time()}).sync()
    print("Published 'hello world'. Staying subscribed for 10 seconds.")
    time.sleep(10)
finally:
    pubnub.unsubscribe().channels('my_channel').execute()
    pubnub.stop()
    print("Unsubscribed and stopped PubNub client.")

view raw JSON →