Logging for Python-Asyncio SDK

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

For an example on how to subscribe with logging, refer to the subscribe example.

Error handling with future()

In case of future() calls errors will be wrapped out by either AsyncioEnvelope or PubNubAsyncioException which both implement two object fields:

• e.result - a request result object in case of success, otherwise None
• e.status - PNStatus object with useful information about the finished request. You may check if it was an error or a success using e.is_error() helper.

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() calls will return you just a result of the request without any state information. Errors should be caught using try/except blocks. For server-side errors a e.status field is populated to provide you more 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