Logging for Python-Asyncio SDK

This page explains how to enable logging and handle errors in the PubNub Python-Asyncio Software Development Kit (SDK). Start by enabling basic logs, then learn two error-handling patterns: future() and result().

How to enable logging

import os
import asyncio
import logging
from pubnub.pnconfiguration import PNConfiguration
from pubnub.pubnub_asyncio import PubNubAsyncio
from pubnub.exceptions import PubNubException


async def demonstrate_logging(pubnub: PubNubAsyncio):
    try:
        # This operation will generate logs we can observe
        envelope = await pubnub.publish() \
            .channel("my_channel") \
            .message({'name': 'Alex', 'action': 'logging_demo'}) \
            .future()

show all 48 lines

To see how logging integrates with subscription operations, refer to the subscribe example. The example sets a stream logger at the DEBUG level and subscribes to a channel using a listener.

Error handling with future()

For future() calls, errors are wrapped in either AsyncioEnvelope or PubNubAsyncioException. Both expose two fields:

• e.result: The request result object if the operation succeeds; otherwise None.
• e.status: A PNStatus object with details about the finished request. Use the e.is_error() helper to determine success or failure.

Future() usage

import asyncio
import os
from pubnub.pnconfiguration import PNConfiguration
from pubnub.pubnub_asyncio import PubNubAsyncio
from pubnub.exceptions import PubNubException


async def publish_future(pubnub: PubNubAsyncio):
    try:
        envelope = await pubnub.publish().channel("my_channel").message("hello!").future()
        if envelope.status.is_error():
            print(f"Error: {envelope.status.error_data}")
            print(f"Error category #{envelope.status.category}")
            return
        else:

show all 41 lines

Error handling with result()

result() returns only the request result, without state information. Catch errors with try/except. For server-side errors, the e.status field provides additional information about the request:

import asyncio
import os
from pubnub.pnconfiguration import PNConfiguration
from pubnub.pubnub_asyncio import PubNubAsyncio
from pubnub.exceptions import PubNubException


async def publish_result(pubnub: PubNubAsyncio):
    try:
        result = await pubnub.publish().channel("my_channel").message("hello!").result()
        print(f"Publish successful: {result}")
    except PubNubException as e:
        print(f"Error: {e}")
        if e.status is not None:
            print(f"Error category #{e.status.category}")

show all 38 lines