Azure Eventhub Py

1---
2name: azure-eventhub-py
3description: |
4  Azure Event Hubs SDK for Python streaming. Use for high-throughput event ingestion, producers, consumers, and checkpointing.
5  Triggers: "event hubs", "EventHubProducerClient", "EventHubConsumerClient", "streaming", "partitions".
6package: azure-eventhub
7---
8 
9# Azure Event Hubs SDK for Python
10 
11Big data streaming platform for high-throughput event ingestion.
12 
13## Installation
14 
15```bash
16pip install azure-eventhub azure-identity
17# For checkpointing with blob storage
18pip install azure-eventhub-checkpointstoreblob-aio
19```
20 
21## Environment Variables
22 
23```bash
24EVENT_HUB_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net
25EVENT_HUB_NAME=my-eventhub
26STORAGE_ACCOUNT_URL=https://<account>.blob.core.windows.net
27CHECKPOINT_CONTAINER=checkpoints
28```
29 
30## Authentication
31 
32```python
33from azure.identity import DefaultAzureCredential
34from azure.eventhub import EventHubProducerClient, EventHubConsumerClient
35 
36credential = DefaultAzureCredential()
37namespace = "<namespace>.servicebus.windows.net"
38eventhub_name = "my-eventhub"
39 
40# Producer
41producer = EventHubProducerClient(
42    fully_qualified_namespace=namespace,
43    eventhub_name=eventhub_name,
44    credential=credential
45)
46 
47# Consumer
48consumer = EventHubConsumerClient(
49    fully_qualified_namespace=namespace,
50    eventhub_name=eventhub_name,
51    consumer_group="$Default",
52    credential=credential
53)
54```
55 
56## Client Types
57 
58| Client | Purpose |
59|--------|---------|
60| `EventHubProducerClient` | Send events to Event Hub |
61| `EventHubConsumerClient` | Receive events from Event Hub |
62| `BlobCheckpointStore` | Track consumer progress |
63 
64## Send Events
65 
66```python
67from azure.eventhub import EventHubProducerClient, EventData
68from azure.identity import DefaultAzureCredential
69 
70producer = EventHubProducerClient(
71    fully_qualified_namespace="<namespace>.servicebus.windows.net",
72    eventhub_name="my-eventhub",
73    credential=DefaultAzureCredential()
74)
75 
76with producer:
77    # Create batch (handles size limits)
78    event_data_batch = producer.create_batch()
79    
80    for i in range(10):
81        try:
82            event_data_batch.add(EventData(f"Event {i}"))
83        except ValueError:
84            # Batch is full, send and create new one
85            producer.send_batch(event_data_batch)
86            event_data_batch = producer.create_batch()
87            event_data_batch.add(EventData(f"Event {i}"))
88    
89    # Send remaining
90    producer.send_batch(event_data_batch)
91```
92 
93### Send to Specific Partition
94 
95```python
96# By partition ID
97event_data_batch = producer.create_batch(partition_id="0")
98 
99# By partition key (consistent hashing)
100event_data_batch = producer.create_batch(partition_key="user-123")
101```
102 
103## Receive Events
104 
105### Simple Receive
106 
107```python
108from azure.eventhub import EventHubConsumerClient
109 
110def on_event(partition_context, event):
111    print(f"Partition: {partition_context.partition_id}")
112    print(f"Data: {event.body_as_str()}")
113    partition_context.update_checkpoint(event)
114 
115consumer = EventHubConsumerClient(
116    fully_qualified_namespace="<namespace>.servicebus.windows.net",
117    eventhub_name="my-eventhub",
118    consumer_group="$Default",
119    credential=DefaultAzureCredential()
120)
121 
122with consumer:
123    consumer.receive(
124        on_event=on_event,
125        starting_position="-1",  # Beginning of stream
126    )
127```
128 
129### With Blob Checkpoint Store (Production)
130 
131```python
132from azure.eventhub import EventHubConsumerClient
133from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
134from azure.identity import DefaultAzureCredential
135 
136checkpoint_store = BlobCheckpointStore(
137    blob_account_url="https://<account>.blob.core.windows.net",
138    container_name="checkpoints",
139    credential=DefaultAzureCredential()
140)
141 
142consumer = EventHubConsumerClient(
143    fully_qualified_namespace="<namespace>.servicebus.windows.net",
144    eventhub_name="my-eventhub",
145    consumer_group="$Default",
146    credential=DefaultAzureCredential(),
147    checkpoint_store=checkpoint_store
148)
149 
150def on_event(partition_context, event):
151    print(f"Received: {event.body_as_str()}")
152    # Checkpoint after processing
153    partition_context.update_checkpoint(event)
154 
155with consumer:
156    consumer.receive(on_event=on_event)
157```
158 
159## Async Client
160 
161```python
162from azure.eventhub.aio import EventHubProducerClient, EventHubConsumerClient
163from azure.identity.aio import DefaultAzureCredential
164import asyncio
165 
166async def send_events():
167    credential = DefaultAzureCredential()
168    
169    async with EventHubProducerClient(
170        fully_qualified_namespace="<namespace>.servicebus.windows.net",
171        eventhub_name="my-eventhub",
172        credential=credential
173    ) as producer:
174        batch = await producer.create_batch()
175        batch.add(EventData("Async event"))
176        await producer.send_batch(batch)
177 
178async def receive_events():
179    async def on_event(partition_context, event):
180        print(event.body_as_str())
181        await partition_context.update_checkpoint(event)
182    
183    async with EventHubConsumerClient(
184        fully_qualified_namespace="<namespace>.servicebus.windows.net",
185        eventhub_name="my-eventhub",
186        consumer_group="$Default",
187        credential=DefaultAzureCredential()
188    ) as consumer:
189        await consumer.receive(on_event=on_event)
190 
191asyncio.run(send_events())
192```
193 
194## Event Properties
195 
196```python
197event = EventData("My event body")
198 
199# Set properties
200event.properties = {"custom_property": "value"}
201event.content_type = "application/json"
202 
203# Read properties (on receive)
204print(event.body_as_str())
205print(event.sequence_number)
206print(event.offset)
207print(event.enqueued_time)
208print(event.partition_key)
209```
210 
211## Get Event Hub Info
212 
213```python
214with producer:
215    info = producer.get_eventhub_properties()
216    print(f"Name: {info['name']}")
217    print(f"Partitions: {info['partition_ids']}")
218    
219    for partition_id in info['partition_ids']:
220        partition_info = producer.get_partition_properties(partition_id)
221        print(f"Partition {partition_id}: {partition_info['last_enqueued_sequence_number']}")
222```
223 
224## Best Practices
225 
2261. **Use batches** for sending multiple events
2272. **Use checkpoint store** in production for reliable processing
2283. **Use async client** for high-throughput scenarios
2294. **Use partition keys** for ordered delivery within a partition
2305. **Handle batch size limits** — catch ValueError when batch is full
2316. **Use context managers** (`with`/`async with`) for proper cleanup
2327. **Set appropriate consumer groups** for different applications
233 
234## Reference Files
235 
236| File | Contents |
237|------|----------|
238| [references/checkpointing.md](references/checkpointing.md) | Checkpoint store patterns, blob checkpointing, checkpoint strategies |
239| [references/partitions.md](references/partitions.md) | Partition management, load balancing, starting positions |
240| [scripts/setup_consumer.py](scripts/setup_consumer.py) | CLI for Event Hub info, consumer setup, and event sending/receiving |
241