PubNub Python SDK
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
-
TypeError: Object of type <YourCustomClass> is not JSON serializable
cause Attempting to publish a Python object that cannot be directly converted to JSON (e.g., a custom class instance, a set) without explicit serialization.fixBefore publishing, ensure your message data is JSON-serializable. Convert custom objects to dictionaries or lists using a custom serializer, or pass only basic Python types (strings, numbers, dicts, lists, booleans, None). -
AttributeError: 'NoneType' object has no attribute 'error_data'
cause This typically occurs in older versions (<10.6.2) when a `PubNubAsyncioException` is raised but the underlying `PNStatus` or `error_data` was not properly set, leading to attempts to access attributes on a `None` object.fixUpgrade your PubNub Python SDK to version 10.6.2 or newer. This version includes fixes to ensure `PubNubAsyncioException` always contains valid error data. -
PubNubException: Not subscribed.
cause This error occurs if you try to perform operations like `publish` or `signal` before the client has successfully established a subscription connection. The client needs to connect and be ready to send/receive.fixWait for a successful `PNConnectedCategory` status in your `SubscribeCallback` before performing publish or other interactive operations. For critical actions, consider using the `pnub.subscribe().sync()` or waiting for the connection status to change.
Warnings
- breaking FCM (Firebase Cloud Messaging) is now the supported push type, replacing GCM (Google Cloud Messaging) and MPNS (Microsoft Push Notification Service), which are deprecated and removed. Applications relying on GCM/MPNS for push notifications must migrate to FCM.
- gotcha Publishing non-JSON-serializable objects (e.g., custom Python classes, sets) prior to v10.6.1 could silently fail without raising an explicit error, leading to messages not being delivered or malformed.
- gotcha In asynchronous contexts (e.g., asyncio), `PubNubAsyncioException` might return without valid `PNStatus` or `error_data`, leading to `AttributeError` when trying to access these properties during error handling.
- breaking Major version changes (e.g., v4 to v7, v7 to v9/v10) often involve significant API refactoring, especially around asynchronous operations and listener patterns. Direct upgrades across major versions without consulting migration guides are likely to break existing code.
Install
-
pip install pubnub
Imports
- PubNub
from pubnub.pubnub import PubNub
- PNConfiguration
from pubnub.pnconfiguration import PNConfiguration
- SubscribeCallback
from pubnub.callbacks import SubscribeCallback
- PNStatusCategory
from pubnub.enums import PNStatusCategory
Quickstart
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.")