Labsco
microsoft logo

azure-eventhub-py

✓ Official★ 2,700

by microsoft · part of microsoft/skills

Big data streaming platform for high-throughput event ingestion.

🔥🔥🔥🔥✓ VerifiedFreeQuick setup
#cloud#git
🧩 One of 7 skills in the microsoft/skills package — works on its own, and pairs well with its siblings.

Big data streaming platform for high-throughput event ingestion.

Inspect the full instructions your agent will receiveExpand

This is the exact playbook injected into your agent when the skill activates — shown here so you can audit it before installing. You don't need to read it to use the skill.

by microsoft

Big data streaming platform for high-throughput event ingestion. npx skills add https://github.com/microsoft/skills --skill azure-eventhub-py Download ZIPGitHub2.7k

Azure Event Hubs SDK for Python

Big data streaming platform for high-throughput event ingestion.

Environment Variables

Copy & paste — that's it
EVENT_HUB_FULLY_QUALIFIED_NAMESPACE= .servicebus.windows.net # Required for all auth methods
EVENT_HUB_NAME=my-eventhub # Required for all auth methods
STORAGE_ACCOUNT_URL=https:// .blob.core.windows.net # Required for checkpoint storage
CHECKPOINT_CONTAINER=checkpoints # Required for checkpoint storage
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication & Lifecycle

🔑 Two rules apply to every code sample below:

  • Prefer DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.

  • Local dev: DefaultAzureCredential works as-is.

  • Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.

  • Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:

  • Sync: with <Client>(...) as client:

  • Async: async with <Client>(...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)

Snippets may abbreviate this setup, but production code should always follow both rules.

Copy & paste — that's it
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.eventhub import EventHubProducerClient, EventHubConsumerClient

# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS= 
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
namespace = " .servicebus.windows.net"
eventhub_name = "my-eventhub"

# Producer
with EventHubProducerClient(
 fully_qualified_namespace=namespace,
 eventhub_name=eventhub_name,
 credential=credential
) as producer:
 # Use producer here (see following sections for operations)
 ...

# Consumer
with EventHubConsumerClient(
 fully_qualified_namespace=namespace,
 eventhub_name=eventhub_name,
 consumer_group="$Default",
 credential=credential
) as consumer:
 # Use consumer here (see following sections for operations)
 ...

Client Types

Client Purpose EventHubProducerClient Send events to Event Hub EventHubConsumerClient Receive events from Event Hub BlobCheckpointStore Track consumer progress

Send Events

Copy & paste — that's it
from azure.eventhub import EventHubProducerClient, EventData
from azure.identity import DefaultAzureCredential

with EventHubProducerClient(
 fully_qualified_namespace=" .servicebus.windows.net",
 eventhub_name="my-eventhub",
 credential=DefaultAzureCredential()
) as producer:
 # Create batch (handles size limits)
 event_data_batch = producer.create_batch()
 
 for i in range(10):
 try:
 event_data_batch.add(EventData(f"Event {i}"))
 except ValueError:
 # Batch is full, send and create new one
 producer.send_batch(event_data_batch)
 event_data_batch = producer.create_batch()
 event_data_batch.add(EventData(f"Event {i}"))
 
 # Send remaining
 producer.send_batch(event_data_batch)

Send to Specific Partition

Copy & paste — that's it
# By partition ID
event_data_batch = producer.create_batch(partition_id="0")

# By partition key (consistent hashing)
event_data_batch = producer.create_batch(partition_key="user-123")

Receive Events

Simple Receive

Copy & paste — that's it
from azure.eventhub import EventHubConsumerClient

def on_event(partition_context, event):
 print(f"Partition: {partition_context.partition_id}")
 print(f"Data: {event.body_as_str()}")
 partition_context.update_checkpoint(event)

with EventHubConsumerClient(
 fully_qualified_namespace=" .servicebus.windows.net",
 eventhub_name="my-eventhub",
 consumer_group="$Default",
 credential=DefaultAzureCredential()
) as consumer:
 consumer.receive(
 on_event=on_event,
 starting_position="-1", # Beginning of stream
 )

With Blob Checkpoint Store (Production)

Copy & paste — that's it
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
from azure.identity import DefaultAzureCredential

checkpoint_store = BlobCheckpointStore(
 blob_account_url="https:// .blob.core.windows.net",
 container_name="checkpoints",
 credential=DefaultAzureCredential()
)

with EventHubConsumerClient(
 fully_qualified_namespace=" .servicebus.windows.net",
 eventhub_name="my-eventhub",
 consumer_group="$Default",
 credential=DefaultAzureCredential(),
 checkpoint_store=checkpoint_store
) as consumer:
 def on_event(partition_context, event):
 print(f"Received: {event.body_as_str()}")
 # Checkpoint after processing
 partition_context.update_checkpoint(event)

 consumer.receive(on_event=on_event)

Async Client

Copy & paste — that's it
from azure.eventhub.aio import EventHubProducerClient, EventHubConsumerClient
from azure.identity.aio import DefaultAzureCredential
import asyncio

async def send_events():
 credential = DefaultAzureCredential()
 
 async with EventHubProducerClient(
 fully_qualified_namespace=" .servicebus.windows.net",
 eventhub_name="my-eventhub",
 credential=credential
 ) as producer:
 batch = await producer.create_batch()
 batch.add(EventData("Async event"))
 await producer.send_batch(batch)

async def receive_events():
 async def on_event(partition_context, event):
 print(event.body_as_str())
 await partition_context.update_checkpoint(event)
 
 async with EventHubConsumerClient(
 fully_qualified_namespace=" .servicebus.windows.net",
 eventhub_name="my-eventhub",
 consumer_group="$Default",
 credential=DefaultAzureCredential()
 ) as consumer:
 await consumer.receive(on_event=on_event)

asyncio.run(send_events())

Event Properties

Copy & paste — that's it
event = EventData("My event body")

# Set properties
event.properties = {"custom_property": "value"}
event.content_type = "application/json"

# Read properties (on receive)
print(event.body_as_str())
print(event.sequence_number)
print(event.offset)
print(event.enqueued_time)
print(event.partition_key)

Get Event Hub Info

Copy & paste — that's it
with producer:
 info = producer.get_eventhub_properties()
 print(f"Name: {info['name']}")
 print(f"Partitions: {info['partition_ids']}")
 
 for partition_id in info['partition_ids']:
 partition_info = producer.get_partition_properties(partition_id)
 print(f"Partition {partition_id}: {partition_info['last_enqueued_sequence_number']}")

Best Practices

  • Pick sync OR async and stay consistent. Do not mix azure.xxx sync clients with azure.xxx.aio async clients in the same call path. Choose one mode per module.

  • Always use context managers for clients and async credentials. Wrap every client in with Client(...) as client: (sync) or async with Client(...) as client: (async) for proper cleanup. For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.

  • Use DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / API keys when possible).

  • Use batches for sending multiple events

  • Use checkpoint store in production for reliable processing

  • Use async client for high-throughput scenarios

  • Use partition keys for ordered delivery within a partition

  • Handle batch size limits — catch ValueError when batch is full

  • Set appropriate consumer groups for different applications

Reference Files

File Contents references/checkpointing.md Checkpoint store patterns, blob checkpointing, checkpoint strategies references/partitions.md Partition management, load balancing, starting positions scripts/setup_consumer.py CLI for Event Hub info, consumer setup, and event sending/receiving