Azure Eventhub Py

from azure.eventhub import EventHubProducerClient
from azure.identity import DefaultAzureCredential

producer = EventHubProducerClient(
    fully_qualified_namespace="<namespace>.servicebus.windows.net",
    eventhub_name="my-eventhub",
    credential=DefaultAzureCredential()
)

with producer:
    # Get Event Hub properties
    eh_props = producer.get_eventhub_properties()
    print(f"Event Hub: {eh_props['name']}")
    print(f"Partitions: {eh_props['partition_ids']}")
    print(f"Created: {eh_props['created_at']}")
    
    # Get individual partition properties
    for partition_id in eh_props['partition_ids']:
        props = producer.get_partition_properties(partition_id)
        print(f"\nPartition {partition_id}:")
        print(f"  First sequence: {props['beginning_sequence_number']}")
        print(f"  Last sequence: {props['last_enqueued_sequence_number']}")
        print(f"  Last offset: {props['last_enqueued_offset']}")
        print(f"  Last enqueued: {props['last_enqueued_time_utc']}")
        print(f"  Is empty: {props['is_empty']}")

# Send to specific partition
batch = producer.create_batch(partition_id="0")
batch.add(EventData("Goes to partition 0"))
producer.send_batch(batch)

# Same key always goes to same partition
batch = producer.create_batch(partition_key="user-123")
batch.add(EventData("All user-123 events in same partition"))
producer.send_batch(batch)

# Different keys may go to different partitions
for user_id in ["user-1", "user-2", "user-3"]:
    batch = producer.create_batch(partition_key=user_id)
    batch.add(EventData(f"Event for {user_id}"))
    producer.send_batch(batch)

# No partition_id or partition_key = round-robin distribution
batch = producer.create_batch()
batch.add(EventData("Distributed across partitions"))
producer.send_batch(batch)

from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore

checkpoint_store = BlobCheckpointStore(...)

consumer = EventHubConsumerClient(
    fully_qualified_namespace="<namespace>.servicebus.windows.net",
    eventhub_name="my-eventhub",
    consumer_group="$Default",
    credential=DefaultAzureCredential(),
    checkpoint_store=checkpoint_store  # Required for load balancing
)

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

async with consumer:
    # Automatically distributes partitions across consumers
    await consumer.receive(on_event=on_event)

async with consumer:
    await consumer.receive(
        on_event=on_event,
        partition_id="0"  # Only receive from partition 0
    )

Consumer 1: Partitions 0, 1, 2
Consumer 2: Partitions 3, 4, 5  
Consumer 3: Partitions 6, 7

async def on_partition_initialize(partition_context):
    """Called when consumer claims a partition."""
    print(f"Initialized partition {partition_context.partition_id}")

async def on_partition_close(partition_context, reason):
    """Called when consumer releases a partition."""
    print(f"Closed partition {partition_context.partition_id}: {reason}")

async with consumer:
    await consumer.receive(
        on_event=on_event,
        on_partition_initialize=on_partition_initialize,
        on_partition_close=on_partition_close
    )

async def on_event(partition_context, event):
    # Partition info
    print(f"Partition ID: {partition_context.partition_id}")
    print(f"Consumer Group: {partition_context.consumer_group}")
    print(f"Event Hub: {partition_context.eventhub_name}")
    print(f"Namespace: {partition_context.fully_qualified_namespace}")
    
    # Last enqueued event (if tracking enabled)
    if partition_context.last_enqueued_event_properties:
        props = partition_context.last_enqueued_event_properties
        print(f"Last sequence: {props.sequence_number}")
        print(f"Last offset: {props.offset}")
        print(f"Last enqueued: {props.enqueued_time}")

async def on_event(partition_context, event):
    props = partition_context.last_enqueued_event_properties
    
    if props:
        # Calculate lag
        current_seq = event.sequence_number
        last_seq = props.sequence_number
        lag = last_seq - current_seq
        
        if lag > 1000:
            print(f"WARNING: Partition {partition_context.partition_id} lag: {lag}")

import asyncio
from azure.eventhub.aio import EventHubConsumerClient

async def process_partition(consumer, partition_id: str):
    """Process a single partition."""
    async def on_event(partition_context, event):
        await process_event(event)
        await partition_context.update_checkpoint(event)
    
    await consumer.receive(
        on_event=on_event,
        partition_id=partition_id
    )

async def parallel_consume():
    consumer = EventHubConsumerClient(...)
    
    async with consumer:
        # Get all partitions
        props = await consumer.get_eventhub_properties()
        
        # Start parallel tasks for each partition
        tasks = [
            process_partition(consumer, pid)
            for pid in props['partition_ids']
        ]
        
        await asyncio.gather(*tasks)

class PartitionBatcher:
    """Batch events by partition for efficient sending."""
    
    def __init__(self, producer):
        self.producer = producer
        self.batches: dict[str, any] = {}
    
    async def add(self, event: EventData, partition_key: str):
        if partition_key not in self.batches:
            self.batches[partition_key] = await self.producer.create_batch(
                partition_key=partition_key
            )
        
        try:
            self.batches[partition_key].add(event)
        except ValueError:
            # Batch full, send and create new
            await self.producer.send_batch(self.batches[partition_key])
            self.batches[partition_key] = await self.producer.create_batch(
                partition_key=partition_key
            )
            self.batches[partition_key].add(event)
    
    async def flush(self):
        for batch in self.batches.values():
            if batch:
                await self.producer.send_batch(batch)
        self.batches.clear()

from azure.eventhub import EventHubConsumerClient

# From beginning
await consumer.receive(
    on_event=on_event,
    starting_position="-1"  # Beginning of stream
)

# From end (new events only)
await consumer.receive(
    on_event=on_event,
    starting_position="@latest"
)

# From specific offset
await consumer.receive(
    on_event=on_event,
    starting_position="12345"  # Specific offset
)

# From specific time
from datetime import datetime, timezone
start_time = datetime(2024, 1, 1, tzinfo=timezone.utc)
await consumer.receive(
    on_event=on_event,
    starting_position=start_time
)

# Different positions per partition
await consumer.receive(
    on_event=on_event,
    starting_position={
        "0": "-1",      # Partition 0 from beginning
        "1": "@latest", # Partition 1 from end
        "2": "5000"     # Partition 2 from offset 5000
    }
)

class PartitionMonitor:
    def __init__(self):
        self.owned_partitions: set[str] = set()
        self.event_counts: dict[str, int] = {}
    
    async def on_partition_initialize(self, partition_context):
        self.owned_partitions.add(partition_context.partition_id)
        print(f"Now own partitions: {self.owned_partitions}")
    
    async def on_partition_close(self, partition_context, reason):
        self.owned_partitions.discard(partition_context.partition_id)
        print(f"Released {partition_context.partition_id}: {reason}")
    
    async def on_event(self, partition_context, event):
        pid = partition_context.partition_id
        self.event_counts[pid] = self.event_counts.get(pid, 0) + 1
        
        # Log distribution every 1000 events
        total = sum(self.event_counts.values())
        if total % 1000 == 0:
            print(f"Event distribution: {self.event_counts}")