Merge pull request #29 from ks6088ts-labs/feature/issue-28_queue-storage

add queue storage APIs

Author: ks6088ts

azure_storage_queue.env.sample

@@ -0,0 +1 @@
+AZURE_STORAGE_QUEUE_CONNECTION_STRING = "<connection-string>"

backend/fastapi.py

@@ -5,6 +5,7 @@
 from backend.routers import azure_event_grid as azure_event_grid_router
 from backend.routers import azure_openai as azure_openai_router
 from backend.routers import azure_storage as azure_storage_router
+from backend.routers import azure_storage_queue as azure_storage_queue_router
 from backend.routers import document_intelligence as document_intelligence_router
 
 app = FastAPI(
@@ -16,6 +17,7 @@
 app.include_router(azure_storage_router.router)
 app.include_router(azure_ai_vision_router.router)
 app.include_router(azure_event_grid_router.router)
+app.include_router(azure_storage_queue_router.router)
 
 
 def custom_openapi():

backend/internals/azure_storage_queue.py

@@ -0,0 +1,75 @@
+from logging import getLogger
+
+from azure.core.paging import ItemPaged
+from azure.storage.queue import QueueMessage, QueueServiceClient
+
+from backend.settings.azure_storage_queue import AzureStorageQueueSettings
+
+logger = getLogger(__name__)
+
+
+class AzureStorageQueueClient:
+    def __init__(self, settings: AzureStorageQueueSettings):
+        self.client = QueueServiceClient.from_connection_string(settings.azure_storage_queue_connection_string)
+
+    def create_queue(
+        self,
+        queue_name: str,
+    ):
+        queue_client = self.client.get_queue_client(queue_name)
+        queue_client.create_queue()
+        logger.info(f"Created queue {queue_name}")
+
+    def delete_queue(
+        self,
+        queue_name: str,
+    ):
+        queue_client = self.client.get_queue_client(queue_name)
+        queue_client.delete_queue()
+        logger.info(f"Deleted queue {queue_name}")
+
+    def send_message(
+        self,
+        queue_name: str,
+        message: str,
+    ) -> QueueMessage:
+        queue_client = self.client.get_queue_client(queue_name)
+        sent_message = queue_client.send_message(
+            content=message,
+        )
+        logger.info(f"Sent message to queue {queue_name}")
+        return sent_message
+
+    def get_queue_properties(
+        self,
+        queue_name: str,
+    ):
+        queue_client = self.client.get_queue_client(queue_name)
+        properties = queue_client.get_queue_properties()
+        logger.info(f"Got properties of queue {queue_name}")
+        return properties
+
+    def receive_messages(
+        self,
+        queue_name: str,
+        max_messages: int = 1,
+    ) -> ItemPaged[QueueMessage]:
+        queue_client = self.client.get_queue_client(queue_name)
+        messages = queue_client.receive_messages(
+            max_messages=max_messages,
+        )
+        logger.info(f"Received messages from queue {queue_name}")
+        return messages
+
+    def delete_message(
+        self,
+        queue_name: str,
+        message_id: str,
+        pop_receipt: str,
+    ):
+        queue_client = self.client.get_queue_client(queue_name)
+        queue_client.delete_message(
+            message=message_id,
+            pop_receipt=pop_receipt,
+        )
+        logger.info(f"Deleted message from queue {queue_name}")

backend/routers/azure_storage_queue.py

@@ -0,0 +1,130 @@
+from logging import getLogger
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+from backend.internals.azure_storage_queue import AzureStorageQueueClient
+from backend.schemas import azure_storage_queue as azure_storage_queue_schemas
+from backend.settings.azure_storage_queue import AzureStorageQueueSettings
+
+logger = getLogger(__name__)
+client = AzureStorageQueueClient(
+    settings=AzureStorageQueueSettings(),
+)
+
+router = APIRouter(
+    prefix="/azure_storage_queue",
+    tags=["azure_storage_queue"],
+    responses={404: {"description": "Not found"}},
+)
+
+
+@router.post(
+    "/queues/",
+    response_model=azure_storage_queue_schemas.CreateQueueResponse,
+    status_code=200,
+)
+async def create_queue(
+    body: azure_storage_queue_schemas.CreateQueueRequest,
+):
+    try:
+        client.create_queue(
+            queue_name=body.queue_name,
+        )
+    except Exception as e:
+        logger.error(f"Failed to create queue: {e}")
+        raise
+    return azure_storage_queue_schemas.CreateQueueResponse(
+        queue_name=body.queue_name,
+    )
+
+
+@router.delete(
+    "/queues/",
+    response_model=azure_storage_queue_schemas.DeleteQueueResponse,
+    status_code=200,
+)
+async def delete_queue(
+    body: azure_storage_queue_schemas.DeleteQueueRequest,
+):
+    try:
+        client.delete_queue(
+            queue_name=body.queue_name,
+        )
+    except Exception as e:
+        logger.error(f"Failed to delete queue: {e}")
+        raise
+    return azure_storage_queue_schemas.DeleteQueueResponse(
+        queue_name=body.queue_name,
+    )
+
+
+@router.post(
+    "/messages/",
+    response_model=azure_storage_queue_schemas.SendMessageResponse,
+    status_code=200,
+)
+async def send_message(
+    body: azure_storage_queue_schemas.SendMessageRequest,
+):
+    try:
+        sent_message = client.send_message(
+            queue_name=body.queue_name,
+            message=body.message,
+        )
+        logger.info(f"Sent message: {sent_message}")
+    except Exception as e:
+        logger.error(f"Failed to send message: {e}")
+        raise
+    return azure_storage_queue_schemas.SendMessageResponse()
+
+
+@router.get(
+    "/messages/",
+    status_code=200,
+)
+async def receive_messages(
+    queue_name: str,
+    max_messages: int = 1,
+):
+    try:
+        messages = client.receive_messages(
+            queue_name=queue_name,
+            max_messages=max_messages,
+        )
+        logger.info(f"Received messages: {messages}")
+        message_list = []
+        for message in messages:
+            message_list.append(
+                {
+                    "id": message.id,
+                    "message": message.content,
+                    "pop_receipt": message.pop_receipt,
+                }
+            )
+    except Exception as e:
+        logger.error(f"Failed to receive messages: {e}")
+        raise
+    return JSONResponse(
+        content=message_list,
+    )
+
+
+@router.delete(
+    "/messages/",
+    response_model=azure_storage_queue_schemas.DeleteMessageResponse,
+    status_code=200,
+)
+async def delete_message(
+    body: azure_storage_queue_schemas.DeleteMessageRequest,
+):
+    try:
+        client.delete_message(
+            queue_name=body.queue_name,
+            message_id=body.message_id,
+            pop_receipt=body.pop_receipt,
+        )
+    except Exception as e:
+        logger.error(f"Failed to delete message: {e}")
+        raise
+    return azure_storage_queue_schemas.DeleteMessageResponse()

backend/schemas/azure_storage_queue.py

@@ -0,0 +1,40 @@
+from logging import getLogger
+
+from pydantic import BaseModel
+
+logger = getLogger(__name__)
+
+
+class CreateQueueRequest(BaseModel):
+    queue_name: str
+
+
+class CreateQueueResponse(BaseModel):
+    queue_name: str
+
+
+class DeleteQueueRequest(BaseModel):
+    queue_name: str
+
+
+class DeleteQueueResponse(BaseModel):
+    queue_name: str
+
+
+class SendMessageRequest(BaseModel):
+    queue_name: str
+    message: str
+
+
+class SendMessageResponse(BaseModel):
+    pass
+
+
+class DeleteMessageRequest(BaseModel):
+    queue_name: str
+    message_id: str
+    pop_receipt: str
+
+
+class DeleteMessageResponse(BaseModel):
+    pass

backend/settings/azure_storage_queue.py

@@ -0,0 +1,10 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class AzureStorageQueueSettings(BaseSettings):
+    azure_storage_queue_connection_string: str = "<connection-string>"
+
+    model_config = SettingsConfigDict(
+        env_file="azure_storage_queue.env",
+        env_file_encoding="utf-8",
+    )

docs/README.md

@@ -14,6 +14,7 @@
 - [FastAPI > UploadFile class](https://fastapi.tiangolo.com/reference/uploadfile/)
 - [FastAPI > Extending OpenAPI](https://fastapi.tiangolo.com/how-to/extending-openapi/)
 - [Get started with Azure Blob Storage and Python](https://learn.microsoft.com/en-us/azure/storage/blobs/storage-blob-python-get-started?tabs=sas-token)
+- [Quickstart: Azure Queue Storage client library for Python](https://learn.microsoft.com/en-us/azure/storage/queues/storage-quickstart-queues-python?tabs=passwordless%2Croles-azure-portal%2Cenvironment-variable-windows%2Csign-in-azure-cli)
 - [Azure Event Grid client library for Python - version 4.19.0](https://learn.microsoft.com/en-us/python/api/overview/azure/eventgrid-readme?view=azure-python)
 - [Azure Event Grid Client Library Python Samples](https://learn.microsoft.com/en-us/samples/azure/azure-sdk-for-python/eventgrid-samples/)
 

poetry.lock

@@ -32,6 +32,7 @@ python-multipart = "^0.0.9"
 azure-storage-blob = "^12.19.1"
 azure-ai-vision-imageanalysis = "^1.0.0b2"
 azure-eventgrid = "^4.19.0"
+azure-storage-queue = "^12.9.0"
 
 
 [tool.poetry.group.frontend.dependencies]