AIO Azure Clients Toolbox

High-performance async Python library for Azure SDK clients with intelligent connection pooling.

Features

• Async apps: Built for high-concurrency async applications: we have used this in production at Mulligan Funding for a few years.
• 20-100x Performance Improvement: Connection pooling reduces operation latency for some services from 100-900ms to 1-5ms.
• Intelligent Connection Management: Automatic lifecycle management with semaphore-based client limiting.
• Azure SDK Integration: Wrappers for Cosmos DB, EventHub, Service Bus, Blob Storage, and EventGrid.
• Testing Utilities: Includes pytest fixtures for mocking Azure services.

Useful Docs

• 📖 Full Docs - Complete guide with examples and API reference.
• Quick Start Guide
• Connection Pooling Deep Dive - Technical details with diagrams.

Installation

pip install aio-azure-clients-toolbox

Quick Start

from azure.identity.aio import DefaultAzureCredential
from aio_azure_clients_toolbox import ManagedCosmos

# Traditional approach - slow
cosmos_client = CosmosClient(endpoint, credential)
container = cosmos_client.get_database("db").get_container("container")
await container.create_item({"id": "1"})  # 200ms+ including connection setup

# Connection pooled approach - fast
cosmos_client = ManagedCosmos(
    endpoint="https://your-cosmos.documents.azure.com:443/",
    dbname="your-database",
    container_name="your-container",
    credential=DefaultAzureCredential(),

    # Pool configuration
    client_limit=100,      # Concurrent clients per connection
    max_size=10,           # Maximum connections in pool
    max_idle_seconds=300   # Connection idle timeout
)

async with cosmos_client.get_container_client() as container:
    await container.create_item({"id": "1"})  # 2ms after pool warmup

Supported Azure Services

Testing Support

Also includes, built-in pytest fixtures for easy testing:

# tests/conftest.py
pytest_plugins = [
    "aio_azure_clients_toolbox.testing_utils.fixtures",
]

# Use in your tests
async def test_cosmos_operations(cosmos_insertable, document):
    container_client, set_return = cosmos_insertable
    set_return("success")
    result = await cosmos_client.insert_doc(document)
    assert result == "success"

Full Client Documentation

For detailed examples and advanced usage patterns, see the complete documentation.

Azure BlobStorage

from aio_azure_clients_toolbox import AzureBlobStorageClient

client = AzureBlobStorageClient(
    az_storage_url="https://account.blob.core.windows.net",
    container_name="my-container",
    az_credential=DefaultAzureCredential()
)

# Upload and download with SAS token support
await client.upload_blob("file.txt", b"content")
data = await client.download_blob("file.txt")
sas_url = await client.get_blob_sas_url("file.txt")

CosmosDB

from aio_azure_clients_toolbox import ManagedCosmos

client = ManagedCosmos(
    endpoint="https://your-account.documents.azure.com:443/",
    dbname="your-database",
    container_name="your-container",
    credential=DefaultAzureCredential()
)
# Document operations
async with cosmos.get_container_client() as container:
    # Create document
    document = {"id": "1", "name": "example", "category": "test"}
    result = await container.create_item(body=document)

    # Read document
    item = await container.read_item(item="1", partition_key="test")

EventGrid

from aio_azure_clients_toolbox.clients.eventgrid import EventGridClient, EventGridConfig

client = EventGridClient(
    config=EventGridConfig([
        EventGridTopicConfig("topic1", "https://topic1.azure.net/api/event"),
        EventGridTopicConfig("topic2", "https://topic2.azure.net/api/event"),
    ]),
    async_credential=DefaultAzureCredential()
)

await client.async_emit_event("topic1", "event-type", "subject", {"data": "value"})

EventHub

from aio_azure_clients_toolbox import ManagedAzureEventhubProducer

client = ManagedAzureEventhubProducer(
    eventhub_namespace="my-namespace.servicebus.windows.net",
    eventhub_name="my-hub",
    credential=DefaultAzureCredential()
)

await client.send_event('{"event": "data"}')

Service Bus

from aio_azure_clients_toolbox import ManagedAzureServiceBusSender

client = ManagedAzureServiceBusSender(
    service_bus_namespace="my-namespace.servicebus.windows.net",
    queue_name="my-queue",
    credential=DefaultAzureCredential()
)

await client.send_message("Hello, Service Bus!")

# Receiving messages
async with client.get_receiver() as receiver:
    async for message in receiver:
        await process_message(message)