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

1import os
2import asyncio
3import logging
4from pubnub.pnconfiguration import PNConfiguration
5from pubnub.pubnub_asyncio import PubNubAsyncio
6from pubnub.exceptions import PubNubException
7

8

9async def demonstrate_logging(pubnub: PubNubAsyncio):
10    try:
11        # This operation will generate logs we can observe
12        envelope = await pubnub.publish() \
13            .channel("my_channel") \
14            .message({'name': 'Alex', 'action': 'logging_demo'}) \
15            .future()
16        print(f"Publish timetoken: {envelope.result.timetoken}")
17    except PubNubException as e:
18        print(f"Error: {e}")
19

20

21if __name__ == "__main__":
22    # Set up logging
23    logging.basicConfig(level=logging.INFO)
24    
25    # Enable detailed PubNub DEBUG logs
26    pubnub_logger = logging.getLogger('pubnub')
27    pubnub_logger.setLevel(logging.DEBUG)
28    
29    # Configuration for PubNub instance
30    pn_config = PNConfiguration()
31    pn_config.subscribe_key = os.getenv('SUBSCRIBE_KEY', 'demo')  # Replace 'demo' with your subscribe key from PubNub Admin Portal
32    pn_config.publish_key = os.getenv('PUBLISH_KEY', 'demo')      # Replace 'demo' with your publish key from PubNub Admin Portal
33    pn_config.user_id = "logging_demo_user"
34

35    # Create and configure the event loop
36    loop = asyncio.new_event_loop()
37    asyncio.set_event_loop(loop)
38    
39    # Initialize PubNub with logging enabled
40    pubnub = PubNubAsyncio(pn_config)
41    
42    try:
43        # Run the demo that will generate logs
44        loop.run_until_complete(demonstrate_logging(pubnub))
45    finally:
46        # Clean up resources
47        loop.run_until_complete(pubnub.stop())
48        loop.close()

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

1import asyncio
2import os
3from pubnub.pnconfiguration import PNConfiguration
4from pubnub.pubnub_asyncio import PubNubAsyncio
5from pubnub.exceptions import PubNubException
6

7

8async def publish_future(pubnub: PubNubAsyncio):
9    try:
10        envelope = await pubnub.publish().channel("my_channel").message("hello!").future()
11        if envelope.status.is_error():
12            print(f"Error: {envelope.status.error_data}")
13            print(f"Error category #{envelope.status.category}")
14            return
15        else:
16            print(f"Publish successful: {envelope.result}")
17    except PubNubException as e:
18        print(f"Publish failed: {e}")
19

20

21if __name__ == "__main__":
22    # Configuration for PubNub instance
23    pn_config = PNConfiguration()
24    pn_config.subscribe_key = os.getenv('SUBSCRIBE_KEY', 'demo')  # Replace 'demo' with your subscribe key from PubNub Admin Portal
25    pn_config.publish_key = os.getenv('PUBLISH_KEY', 'demo')      # Replace 'demo' with your publish key from PubNub Admin Portal
26    pn_config.user_id = "publish_demo_user"
27

28    # Create and configure the event loop
29    loop = asyncio.new_event_loop()
30    asyncio.set_event_loop(loop)
31    
32    # Initialize PubNub
33    pubnub = PubNubAsyncio(pn_config)
34    
35    try:
36        # Run the publish operation
37        loop.run_until_complete(publish_future(pubnub))
38    finally:
39        # Clean up resources
40        loop.run_until_complete(pubnub.stop())
41        loop.close()

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:

1import asyncio
2import os
3from pubnub.pnconfiguration import PNConfiguration
4from pubnub.pubnub_asyncio import PubNubAsyncio
5from pubnub.exceptions import PubNubException
6

7

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

17

18if __name__ == "__main__":
19    # Configuration for PubNub instance
20    pn_config = PNConfiguration()
21    pn_config.subscribe_key = os.getenv('SUBSCRIBE_KEY', 'demo')  # Replace 'demo' with your subscribe key from PubNub Admin Portal
22    pn_config.publish_key = os.getenv('PUBLISH_KEY', 'demo')      # Replace 'demo' with your publish key from PubNub Admin Portal
23    pn_config.user_id = "publish_demo_user"
24

25    # Create and configure the event loop
26    loop = asyncio.new_event_loop()
27    asyncio.set_event_loop(loop)
28    
29    # Initialize PubNub
30    pubnub = PubNubAsyncio(pn_config)
31    
32    try:
33        # Run the publish operation
34        loop.run_until_complete(publish_result(pubnub))
35    finally:
36        # Clean up resources
37        loop.run_until_complete(pubnub.stop())
38        loop.close()

show all 38 lines