linter and updates docs

Signed-off-by: Elena Kolevska <[email protected]>

Author: elena-kolevska

dapr/aio/clients/grpc/subscription.py

@@ -120,4 +120,4 @@ def __aiter__(self):
         return self
 
     async def __anext__(self):
-        return await self.next_message()
+        return await self.next_message()

dapr/clients/grpc/subscription.py

@@ -149,4 +149,3 @@ def __iter__(self):
 
     def __next__(self):
         return self.next_message()
-

daprdocs/content/en/python-sdk-docs/python-client.md

@@ -260,8 +260,8 @@ def mytopic_important(event: v1.Event) -> None:
 You can create a streaming subscription to a PubSub topic using either the `subscribe`
 or `subscribe_handler` methods.
 
-The `subscribe` method returns a `Subscription` object, which allows you to pull messages from the
-stream by
+The `subscribe` method returns an iterable `Subscription` object, which allows you to pull messages from the
+stream by using a `for` loop (ex. `for message in subscription`) or by 
 calling the `next_message` method. This will block on the main thread while waiting for messages.
 When done, you should call the close method to terminate the
 subscription and stop receiving messages.
@@ -281,7 +281,7 @@ Here's an example of using the `subscribe` method:
 import time
 
 from dapr.clients import DaprClient
-from dapr.clients.grpc.subscription import StreamInactiveError
+from dapr.clients.grpc.subscription import StreamInactiveError, StreamCancelledError
 
 counter = 0
 
@@ -303,30 +303,35 @@ def main():
         )
 
         try:
-            while counter < 5:
-                try:
-                    message = subscription.next_message()
+            for message in subscription:
+                if message is None:
+                    print('No message received. The stream might have been cancelled.')
+                    continue
 
-                except StreamInactiveError as e:
+                try:
+                    response_status = process_message(message)
+
+                    if response_status == 'success':
+                        subscription.respond_success(message)
+                    elif response_status == 'retry':
+                        subscription.respond_retry(message)
+                    elif response_status == 'drop':
+                        subscription.respond_drop(message)
+
+                    if counter >= 5:
+                        break
+                except StreamInactiveError:
                     print('Stream is inactive. Retrying...')
                     time.sleep(1)
                     continue
-                if message is None:
-                    print('No message received within timeout period.')
-                    continue
-
-                # Process the message
-                response_status = process_message(message)
-
-                if response_status == 'success':
-                    subscription.respond_success(message)
-                elif response_status == 'retry':
-                    subscription.respond_retry(message)
-                elif response_status == 'drop':
-                    subscription.respond_drop(message)
+                except StreamCancelledError:
+                    print('Stream was cancelled')
+                    break
+                except Exception as e:
+                    print(f'Error occurred during message processing: {e}')
 
         finally:
-            print("Closing subscription...")
+            print('Closing subscription...')
             subscription.close()
 
 

examples/pubsub-streaming-async/subscriber.py

@@ -33,8 +33,10 @@ async def main():
         try:
             async for message in subscription:
                 if message is None:
-                    print('No message received within timeout period. '
-                          'The stream might have been cancelled.')
+                    print(
+                        'No message received within timeout period. '
+                        'The stream might have been cancelled.'
+                    )
                     continue
 
                 try:

examples/pubsub-streaming/subscriber.py

@@ -22,6 +22,7 @@ def process_message(message):
     print(f'Processing message: {message.data()} from {message.topic()}...', flush=True)
     return 'success'
 
+
 def main():
     with DaprClient() as client:
         global counter
@@ -66,5 +67,6 @@ def main():
             print('Closing subscription...')
             subscription.close()
 
+
 if __name__ == '__main__':
     main()