Add Azure Queue Storage

libraries/botbuilder-azure/botbuilder/azure/__init__.py

@@ -6,6 +6,7 @@
 # --------------------------------------------------------------------------
 
 from .about import __version__
+from .azure_queue_storage import AzureQueueStorage
 from .cosmosdb_storage import CosmosDbStorage, CosmosDbConfig, CosmosDbKeyEscape
 from .cosmosdb_partitioned_storage import (
     CosmosDbPartitionedStorage,
@@ -14,6 +15,7 @@
 from .blob_storage import BlobStorage, BlobStorageSettings
 
 __all__ = [
+    "AzureQueueStorage",
     "BlobStorage",
     "BlobStorageSettings",
     "CosmosDbStorage",

libraries/botbuilder-azure/botbuilder/azure/azure_queue_storage.py

@@ -0,0 +1,67 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from azure.core.exceptions import ResourceExistsError
+from azure.storage.queue.aio import QueueClient
+from jsonpickle import encode
+
+from botbuilder.core import QueueStorage
+from botbuilder.schema import Activity
+
+
+class AzureQueueStorage(QueueStorage):
+    def __init__(self, queues_storage_connection_string: str, queue_name: str):
+        if not queues_storage_connection_string:
+            raise Exception("queues_storage_connection_string cannot be empty.")
+        if not queue_name:
+            raise Exception("queue_name cannot be empty.")
+
+        self.__queue_client = QueueClient.from_connection_string(
+            queues_storage_connection_string, queue_name
+        )
+
+        self.__initialized = False
+
+    async def _initialize(self):
+        if self.__initialized is False:
+            # This should only happen once - assuming this is a singleton.
+            # There is no `create_queue_if_exists` or `exists` method, so we need to catch the ResourceExistsError.
+            try:
+                await self.__queue_client.create_queue()
+            except ResourceExistsError:
+                pass
+            self.__initialized = True
+        return self.__initialized
+
+    async def queue_activity(
+        self,
+        activity: Activity,
+        visibility_timeout: int = None,
+        time_to_live: int = None,
+    ) -> str:
+        """
+        Enqueues an Activity for later processing. The visibility timeout specifies how long the message should be
+        visible to Dequeue and Peek operations.
+
+        :param activity: The activity to be queued for later processing.
+        :type activity: :class:`botbuilder.schema.Activity`
+        :param visibility_timeout: Visibility timeout in seconds. Optional with a default value of 0.
+            Cannot be larger than 7 days.
+        :type visibility_timeout: int
+        :param time_to_live: Specifies the time-to-live interval for the message in seconds.
+        :type time_to_live: int
+
+        :returns: QueueMessage as a JSON string.
+        :rtype: :class:`azure.storage.queue.QueueMessage`
+        """
+        await self._initialize()
+
+        # Encode the activity as a JSON string.
+        message = encode(activity)
+
+        receipt = await self.__queue_client.send_message(
+            message, visibility_timeout=visibility_timeout, time_to_live=time_to_live
+        )
+
+        # Encode the QueueMessage receipt as a JSON string.
+        return encode(receipt)

libraries/botbuilder-azure/botbuilder/azure/blob_storage.py

@@ -1,3 +1,6 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
 import json
 from typing import Dict, List
 

libraries/botbuilder-azure/setup.py

@@ -7,6 +7,7 @@
 REQUIRES = [
     "azure-cosmos==3.2.0",
     "azure-storage-blob==12.7.0",
+    "azure-storage-queue==12.1.5",
     "botbuilder-schema==4.13.0",
     "botframework-connector==4.13.0",
     "jsonpickle>=1.2,<1.5",

libraries/botbuilder-azure/tests/test_queue_storage.py

@@ -0,0 +1,50 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import unittest
+import aiounittest
+from jsonpickle import decode
+
+from botbuilder.azure import AzureQueueStorage
+
+EMULATOR_RUNNING = False
+
+# This connection string is to connect to local Azure Storage Emulator.
+CONNECTION_STRING = (
+    "AccountName=devstoreaccount1;"
+    "AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr"
+    "/KBHBeksoGMGw==;DefaultEndpointsProtocol=http;"
+    "BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;"
+    "QueueEndpoint=http://127.0.0.1:10001/devstoreaccount1;"
+    "TableEndpoint=http://127.0.0.1:10002/devstoreaccount1;"
+)
+QUEUE_NAME = "queue"
+
+
+class TestAzureQueueStorageConstructor:
+    def test_queue_storage_init_should_error_without_connection_string(self):
+        try:
+            # pylint: disable=no-value-for-parameter
+            AzureQueueStorage()
+        except Exception as error:
+            assert error
+
+    def test_queue_storage_init_should_error_without_queue_name(self):
+        try:
+            # pylint: disable=no-value-for-parameter
+            AzureQueueStorage(queues_storage_connection_string="somestring")
+        except Exception as error:
+            assert error
+
+
+class TestAzureQueueStorage(aiounittest.AsyncTestCase):
+    @unittest.skipIf(not EMULATOR_RUNNING, reason="Needs the emulator to run.")
+    async def test_returns_read_receipt(self):
+        message = {"string": "test", "object": {"string2": "test2"}, "number": 99}
+        queue = AzureQueueStorage(CONNECTION_STRING, QUEUE_NAME)
+
+        receipt = await queue.queue_activity(message)
+        decoded = decode(receipt)
+
+        assert decoded.id is not None
+        assert decode(decoded.content) == message

libraries/botbuilder-core/botbuilder/core/__init__.py

@@ -30,6 +30,7 @@
 from .middleware_set import AnonymousReceiveMiddleware, Middleware, MiddlewareSet
 from .null_telemetry_client import NullTelemetryClient
 from .private_conversation_state import PrivateConversationState
+from .queue_storage import QueueStorage
 from .recognizer import Recognizer
 from .recognizer_result import RecognizerResult, TopIntent
 from .show_typing_middleware import ShowTypingMiddleware
@@ -77,6 +78,7 @@
     "MiddlewareSet",
     "NullTelemetryClient",
     "PrivateConversationState",
+    "QueueStorage",
     "RegisterClassMiddleware",
     "Recognizer",
     "RecognizerResult",

libraries/botbuilder-core/botbuilder/core/queue_storage.py

@@ -0,0 +1,34 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from abc import ABC, abstractmethod
+from botbuilder.schema import Activity
+
+
+class QueueStorage(ABC):
+    """
+    A base class for enqueueing an Activity for later processing.
+    """
+
+    @abstractmethod
+    async def queue_activity(
+        self,
+        activity: Activity,
+        visibility_timeout: int = None,
+        time_to_live: int = None,
+    ) -> str:
+        """
+        Enqueues an Activity for later processing. The visibility timeout specifies how long the message should be
+        visible to Dequeue and Peek operations.
+
+        :param activity: The activity to be queued for later processing.
+        :type activity: :class:`botbuilder.schema.Activity`
+        :param visibility_timeout: Visibility timeout in seconds. Optional with a default value of 0.
+            Cannot be larger than 7 days.
+        :type visibility_timeout: int
+        :param time_to_live: Specifies the time-to-live interval for the message in seconds.
+        :type time_to_live: int
+
+        :returns: String representing the read receipt.
+        """
+        raise NotImplementedError()