azure-eventhub

Name	azure-eventhub JSON
Version	5.15.1 JSON
	download
home_page	None
Summary	Microsoft Azure Event Hubs Client Library for Python
upload_time	2025-10-30 11:28:21
maintainer	None
docs_url	None
author	None
requires_python	>=3.9
license	None
keywords	azure azure sdk
VCS
bugtrack_url
requirements	No requirements were recorded.
Travis-CI	No Travis.
coveralls test coverage

            # Azure Event Hubs client library for Python

Azure Event Hubs is a highly scalable publish-subscribe service that can ingest millions of events per second and stream
them to multiple consumers. This lets you process and analyze the massive amounts of data produced by your connected
devices and applications. Once Event Hubs has collected the data, you can retrieve, transform, and store it by using
any real-time analytics provider or with batching/storage adapters. If you would like to know more about Azure Event Hubs,
you may wish to review: [What is Event Hubs](https://learn.microsoft.com/azure/event-hubs/event-hubs-about)?

The Azure Event Hubs client library allows for publishing and consuming of Azure Event Hubs events and may be used to:

- Emit telemetry about your application for business intelligence and diagnostic purposes.
- Publish facts about the state of your application which interested parties may observe and use as a trigger for taking action.
- Observe interesting operations and interactions happening within your business or other ecosystem, allowing loosely coupled systems to interact without the need to bind them together.
- Receive events from one or more publishers, transform them to better meet the needs of your ecosystem, then publish the transformed events to a new stream for consumers to observe.

[Source code](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/)
| [Package (PyPi)](https://pypi.org/project/azure-eventhub/)
| [Package (Conda)](https://anaconda.org/microsoft/azure-eventhub/)
| [API reference documentation][api_reference]
| [Product documentation](https://learn.microsoft.com/azure/event-hubs/)
| [Samples](https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/eventhub/azure-eventhub/samples)

## Getting started

### Prerequisites

- Python 3.9 or later.
- **Microsoft Azure Subscription:**  To use Azure services, including Azure Event Hubs, you'll need a subscription.
If you do not have an existing Azure account, you may sign up for a free trial or use your MSDN subscriber benefits when you [create an account](https://azure.microsoft.com/free/).

- **Event Hubs namespace with an Event Hub:** To interact with Azure Event Hubs, you'll also need to have a namespace and Event Hub  available.
If you are not familiar with creating Azure resources, you may wish to follow the step-by-step guide
for [creating an Event Hub using the Azure portal](https://learn.microsoft.com/azure/event-hubs/event-hubs-create).
There, you can also find detailed instructions for using the Azure CLI, Azure PowerShell, or Azure Resource Manager (ARM) templates to create an Event Hub.

### Install the package

Install the Azure Event Hubs client library for Python with pip:

```
$ pip install azure-eventhub
```

### Authenticate the client

Interaction with Event Hubs starts with an instance of EventHubConsumerClient or EventHubProducerClient class. You need either the host name, SAS/AAD credential and event hub name or a connection string to instantiate the client object.

**[Create client from connection string:](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/samples/sync_samples/connection_string_authentication.py)**

For the Event Hubs client library to interact with an Event Hub, the easiest means is to use a connection string, which is created automatically when creating an Event Hubs namespace.
If you aren't familiar with shared access policies in Azure, you may wish to follow the step-by-step guide to [get an Event Hubs connection string](https://learn.microsoft.com/azure/event-hubs/event-hubs-get-connection-string).

- The `from_connection_string` method takes the connection string of the form
`Endpoint=sb://<yournamespace>.servicebus.windows.net/;SharedAccessKeyName=<yoursharedaccesskeyname>;SharedAccessKey=<yoursharedaccesskey>` and
entity name to your Event Hub instance. You can get the connection string from the [Azure portal](https://learn.microsoft.com/azure/event-hubs/event-hubs-get-connection-string#get-connection-string-from-the-portal).

**[Create client using the azure-identity library:](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/samples/sync_samples/client_identity_authentication.py)**

Alternately, one can use a Credential object to authenticate via AAD with the azure-identity package.

- This constructor demonstrated in the sample linked above takes the host name and entity name of your Event Hub instance and credential that implements the
[TokenCredential](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/core/azure-core/azure/core/credentials.py)
protocol. There are implementations of the `TokenCredential` protocol available in the
[azure-identity package](https://pypi.org/project/azure-identity/). The host name is of the format `<yournamespace.servicebus.windows.net>`.
- To use the credential types provided by `azure-identity`, please install the package:
```pip install azure-identity```
- Additionally, to use the async API,  you must first install an async transport, such as [`aiohttp`](https://pypi.org/project/aiohttp/):
```pip install aiohttp```
- When using Azure Active Directory, your principal must be assigned a role which allows access to Event Hubs, such as the
Azure Event Hubs Data Owner role. For more information about using Azure Active Directory authorization with Event Hubs,
please refer to [the associated documentation](https://learn.microsoft.com/azure/event-hubs/authorize-access-azure-active-directory).

## Key concepts

- An **EventHubProducerClient** is a source of telemetry data, diagnostics information, usage logs, or other log data,
as part of an embedded device solution, a mobile device application, a game title running on a console or other device,
some client or server based business solution, or a web site.

- An **EventHubConsumerClient** picks up such information from the Event Hub and processes it. Processing may involve aggregation,
complex computation, and filtering. Processing may also involve distribution or storage of the information in a raw or transformed fashion.
Event Hub consumers are often robust and high-scale platform infrastructure parts with built-in analytics capabilities,
like Azure Stream Analytics, Apache Spark, or Apache Storm.

- A **partition** is an ordered sequence of events that is held in an Event Hub. Azure Event Hubs provides message streaming
through a partitioned consumer pattern in which each consumer only reads a specific subset, or partition, of the message stream.
As newer events arrive, they are added to the end of this sequence. The number of partitions is specified at the time an Event Hub is created and cannot be changed.

- A **consumer group** is a view of an entire Event Hub. Consumer groups enable multiple consuming applications to each
have a separate view of the event stream, and to read the stream independently at their own pace and from their own position.
There can be at most 5 concurrent readers on a partition per consumer group; however it is recommended that there is only
one active consumer for a given partition and consumer group pairing. Each active reader receives all of the events from
its partition; if there are multiple readers on the same partition, then they will receive duplicate events.

For more concepts and deeper discussion, see: [Event Hubs Features](https://learn.microsoft.com/azure/event-hubs/event-hubs-features).
Also, the concepts for AMQP are well documented in [OASIS Advanced Messaging Queuing Protocol (AMQP) Version 1.0](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-overview-v1.0-os.html).

### Thread safety

We do not guarantee that the EventHubProducerClient or EventHubConsumerClient are thread-safe or coroutine-safe. We do not recommend reusing these instances across threads or sharing them between coroutines. It is up to the running application to use these classes in a concurrency-safe manner.

The data model type, `EventDataBatch` is not thread-safe or coroutine-safe. It should not be shared across threads nor used concurrently with client methods.

For scenarios requiring concurrent sending from multiple threads, ensure proper thread-safety management using mechanisms like threading.Lock(). **Note:** Native async APIs should be used instead of running in a ThreadPoolExecutor, if possible.
```python
import threading
from concurrent.futures import ThreadPoolExecutor
from azure.eventhub import EventHubProducerClient, EventData
from azure.identity import DefaultAzureCredential

EVENTHUB_NAMESPACE = "<your-namespace>.servicebus.windows.net"
EVENTHUB_NAME = "<your-eventhub-name>"

# Create a global lock
producer_lock = threading.Lock()

def send_batch(producer_id, producer):
    with producer_lock:
        event_data_batch = producer.create_batch()
        for i in range(10):
            event_data_batch.add(EventData(f"Message {i} from producer {producer_id}"))
        producer.send_batch(event_data_batch)
        print(f"Producer {producer_id} sent batch.")

credential = DefaultAzureCredential()
producer = EventHubProducerClient(
    fully_qualified_namespace=EVENTHUB_NAMESPACE,
    eventhub_name=EVENTHUB_NAME,
    credential=credential
)

with producer:
    with ThreadPoolExecutor(max_workers=5) as executor:
        for i in range(5):  # Launch 5 threads
            executor.submit(send_batch, i, producer)
```

For scenarios requiring concurrent sending in asyncio applications, ensure proper coroutine-safety management using mechanisms like asyncio.Lock()
```python
import asyncio
from azure.eventhub.aio import EventHubProducerClient
from azure.eventhub import EventData
from azure.identity.aio import DefaultAzureCredential

EVENTHUB_NAMESPACE = "<your-namespace>.servicebus.windows.net"
EVENTHUB_NAME = "<your-eventhub-name>"

# Shared lock for coroutine-safe access
producer_lock = asyncio.Lock()

async def send_batch(producer_id, producer):
    async with producer_lock:
        event_data_batch = await producer.create_batch()
        for i in range(10):
            event_data_batch.add(EventData(f"Message {i} from producer {producer_id}"))
        await producer.send_batch(event_data_batch)
        print(f"Producer {producer_id} sent batch.")

credential = DefaultAzureCredential()
producer = EventHubProducerClient(
    fully_qualified_namespace=EVENTHUB_NAMESPACE,
    eventhub_name=EVENTHUB_NAME,
    credential=credential
)

async with producer:
    await asyncio.gather(*(send_batch(i, producer) for i in range(5)))
```

## Examples

The following sections provide several code snippets covering some of the most common Event Hubs tasks, including:

- [Inspect an Event Hub](#inspect-an-event-hub)
- [Publish events to an Event Hub](#publish-events-to-an-event-hub)
- [Consume events from an Event Hub](#consume-events-from-an-event-hub)
- [Consume events from an Event Hub in batches](#consume-events-from-an-event-hub-in-batches)
- [Publish events to an Event Hub asynchronously](#publish-events-to-an-event-hub-asynchronously)
- [Consume events from an Event Hub asynchronously](#consume-events-from-an-event-hub-asynchronously)
- [Consume events from an Event Hub in batches asynchronously](#consume-events-from-an-event-hub-in-batches-asynchronously)
- [Consume events and save checkpoints using a checkpoint store](#consume-events-and-save-checkpoints-using-a-checkpoint-store)
- [Use EventHubConsumerClient to work with IoT Hub](#use-eventhubconsumerclient-to-work-with-iot-hub)

### Inspect an Event Hub

Get the partition ids of an Event Hub.

```python
import os
from azure.eventhub import EventHubConsumerClient
from azure.identity import DefaultAzureCredential

FULLY_QUALIFIED_NAMESPACE = os.environ["EVENT_HUB_HOSTNAME"]
EVENTHUB_NAME = os.environ['EVENT_HUB_NAME']

consumer_client = EventHubConsumerClient(
    fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
    consumer_group='$Default',
    eventhub_name=EVENTHUB_NAME,
    credential=DefaultAzureCredential(),
)

with consumer_client:
    pass # consumer_client is now ready to be used.
```

### Publish events to an Event Hub

Use the `create_batch` method on `EventHubProducerClient` to create an `EventDataBatch` object which can then be sent using the `send_batch` method.
Events may be added to the `EventDataBatch` using the `add` method until the maximum batch size limit in bytes has been reached.

<!-- SNIPPET:send.send_event_data_batch -->

```python
def send_event_data_batch(producer):
    # Without specifying partition_id or partition_key
    # the events will be distributed to available partitions via round-robin.
    event_data_batch = producer.create_batch()
    event_data_batch.add(EventData("Single message"))
    producer.send_batch(event_data_batch)
```

<!-- END SNIPPET -->

### Consume events from an Event Hub

There are multiple ways to consume events from an EventHub.  To simply trigger a callback when an event is received,
the `EventHubConsumerClient.receive` method will be of use as follows:

```python
import logging
from azure.eventhub import EventHubConsumerClient
from azure.identity import DefaultAzureCredential

fully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
client = EventHubConsumerClient(
    fully_qualified_namespace=fully_qualified_namespace,
    eventhub_name=eventhub_name,
    consumer_group=consumer_group,
    credential=DefaultAzureCredential(),
)

logger = logging.getLogger("azure.eventhub")
logging.basicConfig(level=logging.INFO)

def on_event(partition_context, event):
    logger.info("Received event from partition {}".format(partition_context.partition_id))
    partition_context.update_checkpoint(event)

with client:
    client.receive(
        on_event=on_event,
        starting_position="-1",  # "-1" is from the beginning of the partition.
    )
    # receive events from specified partition:
    # client.receive(on_event=on_event, partition_id='0')
```

### Consume events from an Event Hub in batches

Whereas the above sample triggers the callback for each message as it is received, the following sample
triggers the callback on a batch of events, attempting to receive a number at a time.

```python
import logging
from azure.eventhub import EventHubConsumerClient
from azure.identity import DefaultAzureCredential

fully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
client = EventHubConsumerClient(
    fully_qualified_namespace=fully_qualified_namespace,
    eventhub_name=eventhub_name,
    consumer_group=consumer_group,
    credential=DefaultAzureCredential(),
)

logger = logging.getLogger("azure.eventhub")
logging.basicConfig(level=logging.INFO)

def on_event_batch(partition_context, events):
    logger.info("Received event from partition {}".format(partition_context.partition_id))
    partition_context.update_checkpoint()

with client:
    client.receive_batch(
        on_event_batch=on_event_batch,
        starting_position="-1",  # "-1" is from the beginning of the partition.
    )
    # receive events from specified partition:
    # client.receive_batch(on_event_batch=on_event_batch, partition_id='0')
```

### Publish events to an Event Hub asynchronously

Use the `create_batch` method on `EventHubProducer` to create an `EventDataBatch` object which can then be sent using the `send_batch` method.
Events may be added to the `EventDataBatch` using the `add` method until the maximum batch size limit in bytes has been reached.
```python
import asyncio
from azure.eventhub.aio import EventHubProducerClient   # The package name suffixed with ".aio" for async
from azure.eventhub import EventData
from azure.identity.aio import DefaultAzureCredential

fully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'

async def create_batch(client):
    event_data_batch = await client.create_batch()
    can_add = True
    while can_add:
        try:
            event_data_batch.add(EventData('Message inside EventBatchData'))
        except ValueError:
            can_add = False  # EventDataBatch object reaches max_size.
    return event_data_batch

async def send():
    client = EventHubProducerClient(
        fully_qualified_namespace=fully_qualified_namespace,
        eventhub_name=eventhub_name,
        credential=DefaultAzureCredential(),
    )
    batch_data = await create_batch(client)
    async with client:
        await client.send_batch(batch_data)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(send())
```

### Consume events from an Event Hub asynchronously

This SDK supports both synchronous and asyncio based code.  To receive as demonstrated in the samples above, but within
aio, one would need the following:

```python
import logging
import asyncio
from azure.eventhub.aio import EventHubConsumerClient
from azure.identity.aio import DefaultAzureCredential

fully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'

logger = logging.getLogger("azure.eventhub")
logging.basicConfig(level=logging.INFO)

async def on_event(partition_context, event):
    logger.info("Received event from partition {}".format(partition_context.partition_id))
    await partition_context.update_checkpoint(event)

async def receive():
    client = EventHubConsumerClient(
        fully_qualified_namespace=fully_qualified_namespace,
        eventhub_name=eventhub_name,
        consumer_group=consumer_group,
        credential=DefaultAzureCredential(),
    )
    async with client:
        await client.receive(
            on_event=on_event,
            starting_position="-1",  # "-1" is from the beginning of the partition.
        )
        # receive events from specified partition:
        # await client.receive(on_event=on_event, partition_id='0')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(receive())
```

### Consume events from an Event Hub in batches asynchronously

All synchronous functions are supported in aio as well.  As demonstrated above for synchronous batch receipt, one can accomplish
the same within asyncio as follows:

```python
import logging
import asyncio
from azure.eventhub.aio import EventHubConsumerClient
from azure.identity.aio import DefaultAzureCredential

fully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'

logger = logging.getLogger("azure.eventhub")
logging.basicConfig(level=logging.INFO)

async def on_event_batch(partition_context, events):
    logger.info("Received event from partition {}".format(partition_context.partition_id))
    await partition_context.update_checkpoint()

async def receive_batch():
    client = EventHubConsumerClient(
        fully_qualified_namespace=fully_qualified_namespace,
        eventhub_name=eventhub_name,
        consumer_group=consumer_group,
        credential=DefaultAzureCredential(),
    )
    async with client:
        await client.receive_batch(
            on_event_batch=on_event_batch,
            starting_position="-1",  # "-1" is from the beginning of the partition.
        )
        # receive events from specified partition:
        # await client.receive_batch(on_event_batch=on_event_batch, partition_id='0')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(receive_batch())
```

### Consume events and save checkpoints using a checkpoint store

`EventHubConsumerClient` is a high level construct which allows you to receive events from multiple partitions at once
and load balance with other consumers using the same Event Hub and consumer group.

This also allows the user to track progress when events are processed using checkpoints.

A checkpoint is meant to represent the last successfully processed event by the user from a particular partition of
a consumer group in an Event Hub instance. The `EventHubConsumerClient` uses an instance of `CheckpointStore` to update checkpoints
and to store the relevant information required by the load balancing algorithm.

Search pypi with the prefix `azure-eventhub-checkpointstore` to
find packages that support this and use the `CheckpointStore` implementation from one such package. Please note that both sync and async libraries are provided.

In the below example, we create an instance of `EventHubConsumerClient` and use a `BlobCheckpointStore`. You need
to [create an Azure Storage account](https://learn.microsoft.com/azure/storage/common/storage-quickstart-create-account?tabs=azure-portal)
and a [Blob Container](https://learn.microsoft.com/azure/storage/blobs/storage-quickstart-blobs-portal#create-a-container) to run the code.

[Azure Blob Storage Checkpoint Store Async](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub-checkpointstoreblob-aio)
and [Azure Blob Storage Checkpoint Store Sync](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub-checkpointstoreblob)
are one of the `CheckpointStore` implementations we provide that applies Azure Blob Storage as the persistent store.


```python
import asyncio

from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
from azure.identity.aio import DefaultAzureCredential

fully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
blob_account_url = '<< STORAGE ACCOUNT URL >>'
container_name = '<<NAME OF THE BLOB CONTAINER>>'

async def on_event(partition_context, event):
    # do something
    await partition_context.update_checkpoint(event)  # Or update_checkpoint every N events for better performance.

async def receive(client):
    await client.receive(
        on_event=on_event,
        starting_position="-1",  # "-1" is from the beginning of the partition.
    )

async def main():
    checkpoint_store = BlobCheckpointStore(
        blob_account_url=blob_account_url,
        container_name=container_name,
        credential=DefaultAzureCredential()
    )
    client = EventHubConsumerClient(
        fully_qualified_namespace=fully_qualified_namespace,
        eventhub_name=eventhub_name,
        credential=DefaultAzureCredential(),
        consumer_group=consumer_group,
        checkpoint_store=checkpoint_store,  # For load balancing and checkpoint. Leave None for no load balancing
    )
    async with client:
        await receive(client)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
```

### Use EventHubConsumerClient to work with IoT Hub

You can use `EventHubConsumerClient` to work with IoT Hub as well. This is useful for receiving telemetry data of IoT Hub from the
linked EventHub. The associated connection string will not have send claims, hence sending events is not possible.

Please notice that the connection string needs to be for an [Event Hub-compatible endpoint](https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-messages-read-builtin),
e.g. "Endpoint=sb://my-iothub-namespace-[uid].servicebus.windows.net/;SharedAccessKeyName=my-SA-name;SharedAccessKey=my-SA-key;EntityPath=my-iot-hub-name"

There are two ways to get the Event Hubs compatible endpoint:
- Manually get the "Built-in endpoints" of the IoT Hub in Azure Portal and receive from it.
```python
from azure.eventhub import EventHubConsumerClient

connection_str = 'Endpoint=sb://my-iothub-namespace-[uid].servicebus.windows.net/;SharedAccessKeyName=my-SA-name;SharedAccessKey=my-SA-key;EntityPath=my-iot-hub-name'
consumer_group = '<< CONSUMER GROUP >>'
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group)

partition_ids = client.get_partition_ids()
```
- Programmatically retrieve the built-in Event Hubs compatible endpoint.
Refer to [IoT Hub Connection String Sample](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/samples/async_samples/iot_hub_connection_string_receive_async.py).

## Troubleshooting

See the `azure-eventhub` [troubleshooting guide](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/TROUBLESHOOTING.md) for details on how to diagnose various failure scenarios.

### Logging

- Enable `azure.eventhub` logger to collect traces from the library.
- Enable AMQP frame level trace by setting `logging_enable=True` when creating the client.
- Refer to [this guide](https://learn.microsoft.com/azure/developer/python/sdk/azure-sdk-logging) on configuring logging for Azure libraries for Python for additional information.

```python
import logging
import sys

handler = logging.StreamHandler(stream=sys.stdout)
log_fmt = logging.Formatter(fmt="%(asctime)s | %(threadName)s | %(levelname)s | %(name)s | %(message)s")
handler.setFormatter(log_fmt)
logger = logging.getLogger('azure.eventhub')
logger.setLevel(logging.DEBUG)
logger.addHandler(handler)

...

from azure.eventhub import EventHubProducerClient, EventHubConsumerClient

producer = EventHubProducerClient(..., logging_enable=True)
consumer = EventHubConsumerClient(..., logging_enable=True)
```

## Next steps

### More sample code

Please take a look at the [samples](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/samples) directory for detailed examples of how to use this library to send and receive events to/from Event Hubs.

### Documentation

Reference documentation is available [here](https://azuresdkdocs.z19.web.core.windows.net/python/azure-eventhub/latest/azure.eventhub.html).

### Schema Registry and Avro Encoder

The EventHubs SDK integrates nicely with the [Schema Registry][schemaregistry_service] service and [Avro][avro].
For more information, please refer to [Schema Registry SDK][schemaregistry_repo] and [Schema Registry Avro Encoder SDK][schemaregistry_avroencoder_repo].

### Pure Python AMQP Transport and Backward Compatibility Support

The Azure Event Hubs client library is now based on a pure Python AMQP implementation. `uAMQP` has been removed as required dependency.

To use `uAMQP` as the underlying transport:

1. Install `uamqp` with pip.

```
$ pip install uamqp 
```

2. Pass `uamqp_transport=True` during client construction.

```python
from azure.eventhub import EventHubProducerClient, EventHubConsumerClient
from azure.identity import DefaultAzureCredential

fully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'

client = EventHubProducerClient(
    fully_qualified_namespace=fully_qualified_namespace,
    eventhub_name=eventhub_name,
    credential=DefaultAzureCredential(),
    uamqp_transport=True
)
client = EventHubConsumerClient(
    fully_qualified_namespace=fully_qualified_namespace,
    eventhub_name=eventhub_name,
    credential=DefaultAzureCredential(),
    consumer_group=consumer_group,
    uamqp_transport=True
)
```

Note: The `message` attribute on `EventData`/`EventDataBatch`, which previously exposed the `uamqp.Message`, has been deprecated.
 The "Legacy" objects returned by `EventData.message`/`EventDataBatch.message` have been introduced to help facilitate the transition.

### Building uAMQP wheel from source

If [uAMQP](https://pypi.org/project/uamqp/) is intended to be used as the underlying AMQP protocol implementation for `azure-eventhub`,
uAMQP wheels can be found for most major operating systems.

If you intend to use `uAMQP` and you're running on a platform for which uAMQP wheels are not provided, please follow
 the [uAMQP Installation](https://github.com/Azure/azure-uamqp-python#installation) guidance to install from source.

### Provide Feedback

If you encounter any bugs or have suggestions, please file an issue in the [Issues](https://github.com/Azure/azure-sdk-for-python/issues) section of the project.

## Contributing

This project welcomes contributions and suggestions.  Most contributions require you to agree to a
Contributor License Agreement (CLA) declaring that you have the right to, and actually do, grant us the rights to use your contribution. For details, visit https://cla.microsoft.com.

When you submit a pull request, a CLA-bot will automatically determine whether you need to provide a CLA and decorate the
PR appropriately (e.g., label, comment). Simply follow the instructions provided by the bot. You will only need to do this once across all repos using our CLA.

This project has adopted the [Microsoft Open Source Code of Conduct](https://opensource.microsoft.com/codeofconduct/).
For more information see the [Code of Conduct FAQ](https://opensource.microsoft.com/codeofconduct/faq/) or contact [opencode@microsoft.com](mailto:opencode@microsoft.com) with any additional questions or comments.

[avro]: https://avro.apache.org/
[api_reference]: https://learn.microsoft.com/python/api/overview/azure/eventhub-readme
[schemaregistry_service]: https://aka.ms/schemaregistry
[schemaregistry_repo]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/schemaregistry/azure-schemaregistry
[schemaregistry_avroencoder_repo]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/schemaregistry/azure-schemaregistry-avroencoder



            

Raw data

            {
    "_id": null,
    "home_page": null,
    "name": "azure-eventhub",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3.9",
    "maintainer_email": null,
    "keywords": "azure, azure sdk",
    "author": null,
    "author_email": "Microsoft Corporation <azpysdkhelp@microsoft.com> License-Expression: MIT",
    "download_url": "https://files.pythonhosted.org/packages/20/bf/96eec5aa57cb08f1e84e1dfa3888e52dbba53e218596cdab5b61d795a1ae/azure_eventhub-5.15.1.tar.gz",
    "platform": null,
    "description": "# Azure Event Hubs client library for Python\n\nAzure Event Hubs is a highly scalable publish-subscribe service that can ingest millions of events per second and stream\nthem to multiple consumers. This lets you process and analyze the massive amounts of data produced by your connected\ndevices and applications. Once Event Hubs has collected the data, you can retrieve, transform, and store it by using\nany real-time analytics provider or with batching/storage adapters. If you would like to know more about Azure Event Hubs,\nyou may wish to review: [What is Event Hubs](https://learn.microsoft.com/azure/event-hubs/event-hubs-about)?\n\nThe Azure Event Hubs client library allows for publishing and consuming of Azure Event Hubs events and may be used to:\n\n- Emit telemetry about your application for business intelligence and diagnostic purposes.\n- Publish facts about the state of your application which interested parties may observe and use as a trigger for taking action.\n- Observe interesting operations and interactions happening within your business or other ecosystem, allowing loosely coupled systems to interact without the need to bind them together.\n- Receive events from one or more publishers, transform them to better meet the needs of your ecosystem, then publish the transformed events to a new stream for consumers to observe.\n\n[Source code](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/)\n| [Package (PyPi)](https://pypi.org/project/azure-eventhub/)\n| [Package (Conda)](https://anaconda.org/microsoft/azure-eventhub/)\n| [API reference documentation][api_reference]\n| [Product documentation](https://learn.microsoft.com/azure/event-hubs/)\n| [Samples](https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/eventhub/azure-eventhub/samples)\n\n## Getting started\n\n### Prerequisites\n\n- Python 3.9 or later.\n- **Microsoft Azure Subscription:**  To use Azure services, including Azure Event Hubs, you'll need a subscription.\nIf you do not have an existing Azure account, you may sign up for a free trial or use your MSDN subscriber benefits when you [create an account](https://azure.microsoft.com/free/).\n\n- **Event Hubs namespace with an Event Hub:** To interact with Azure Event Hubs, you'll also need to have a namespace and Event Hub  available.\nIf you are not familiar with creating Azure resources, you may wish to follow the step-by-step guide\nfor [creating an Event Hub using the Azure portal](https://learn.microsoft.com/azure/event-hubs/event-hubs-create).\nThere, you can also find detailed instructions for using the Azure CLI, Azure PowerShell, or Azure Resource Manager (ARM) templates to create an Event Hub.\n\n### Install the package\n\nInstall the Azure Event Hubs client library for Python with pip:\n\n```\n$ pip install azure-eventhub\n```\n\n### Authenticate the client\n\nInteraction with Event Hubs starts with an instance of EventHubConsumerClient or EventHubProducerClient class. You need either the host name, SAS/AAD credential and event hub name or a connection string to instantiate the client object.\n\n**[Create client from connection string:](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/samples/sync_samples/connection_string_authentication.py)**\n\nFor the Event Hubs client library to interact with an Event Hub, the easiest means is to use a connection string, which is created automatically when creating an Event Hubs namespace.\nIf you aren't familiar with shared access policies in Azure, you may wish to follow the step-by-step guide to [get an Event Hubs connection string](https://learn.microsoft.com/azure/event-hubs/event-hubs-get-connection-string).\n\n- The `from_connection_string` method takes the connection string of the form\n`Endpoint=sb://<yournamespace>.servicebus.windows.net/;SharedAccessKeyName=<yoursharedaccesskeyname>;SharedAccessKey=<yoursharedaccesskey>` and\nentity name to your Event Hub instance. You can get the connection string from the [Azure portal](https://learn.microsoft.com/azure/event-hubs/event-hubs-get-connection-string#get-connection-string-from-the-portal).\n\n**[Create client using the azure-identity library:](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/samples/sync_samples/client_identity_authentication.py)**\n\nAlternately, one can use a Credential object to authenticate via AAD with the azure-identity package.\n\n- This constructor demonstrated in the sample linked above takes the host name and entity name of your Event Hub instance and credential that implements the\n[TokenCredential](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/core/azure-core/azure/core/credentials.py)\nprotocol. There are implementations of the `TokenCredential` protocol available in the\n[azure-identity package](https://pypi.org/project/azure-identity/). The host name is of the format `<yournamespace.servicebus.windows.net>`.\n- To use the credential types provided by `azure-identity`, please install the package:\n```pip install azure-identity```\n- Additionally, to use the async API,  you must first install an async transport, such as [`aiohttp`](https://pypi.org/project/aiohttp/):\n```pip install aiohttp```\n- When using Azure Active Directory, your principal must be assigned a role which allows access to Event Hubs, such as the\nAzure Event Hubs Data Owner role. For more information about using Azure Active Directory authorization with Event Hubs,\nplease refer to [the associated documentation](https://learn.microsoft.com/azure/event-hubs/authorize-access-azure-active-directory).\n\n## Key concepts\n\n- An **EventHubProducerClient** is a source of telemetry data, diagnostics information, usage logs, or other log data,\nas part of an embedded device solution, a mobile device application, a game title running on a console or other device,\nsome client or server based business solution, or a web site.\n\n- An **EventHubConsumerClient** picks up such information from the Event Hub and processes it. Processing may involve aggregation,\ncomplex computation, and filtering. Processing may also involve distribution or storage of the information in a raw or transformed fashion.\nEvent Hub consumers are often robust and high-scale platform infrastructure parts with built-in analytics capabilities,\nlike Azure Stream Analytics, Apache Spark, or Apache Storm.\n\n- A **partition** is an ordered sequence of events that is held in an Event Hub. Azure Event Hubs provides message streaming\nthrough a partitioned consumer pattern in which each consumer only reads a specific subset, or partition, of the message stream.\nAs newer events arrive, they are added to the end of this sequence. The number of partitions is specified at the time an Event Hub is created and cannot be changed.\n\n- A **consumer group** is a view of an entire Event Hub. Consumer groups enable multiple consuming applications to each\nhave a separate view of the event stream, and to read the stream independently at their own pace and from their own position.\nThere can be at most 5 concurrent readers on a partition per consumer group; however it is recommended that there is only\none active consumer for a given partition and consumer group pairing. Each active reader receives all of the events from\nits partition; if there are multiple readers on the same partition, then they will receive duplicate events.\n\nFor more concepts and deeper discussion, see: [Event Hubs Features](https://learn.microsoft.com/azure/event-hubs/event-hubs-features).\nAlso, the concepts for AMQP are well documented in [OASIS Advanced Messaging Queuing Protocol (AMQP) Version 1.0](https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-overview-v1.0-os.html).\n\n### Thread safety\n\nWe do not guarantee that the EventHubProducerClient or EventHubConsumerClient are thread-safe or coroutine-safe. We do not recommend reusing these instances across threads or sharing them between coroutines. It is up to the running application to use these classes in a concurrency-safe manner.\n\nThe data model type, `EventDataBatch` is not thread-safe or coroutine-safe. It should not be shared across threads nor used concurrently with client methods.\n\nFor scenarios requiring concurrent sending from multiple threads, ensure proper thread-safety management using mechanisms like threading.Lock(). **Note:** Native async APIs should be used instead of running in a ThreadPoolExecutor, if possible.\n```python\nimport threading\nfrom concurrent.futures import ThreadPoolExecutor\nfrom azure.eventhub import EventHubProducerClient, EventData\nfrom azure.identity import DefaultAzureCredential\n\nEVENTHUB_NAMESPACE = \"<your-namespace>.servicebus.windows.net\"\nEVENTHUB_NAME = \"<your-eventhub-name>\"\n\n# Create a global lock\nproducer_lock = threading.Lock()\n\ndef send_batch(producer_id, producer):\n    with producer_lock:\n        event_data_batch = producer.create_batch()\n        for i in range(10):\n            event_data_batch.add(EventData(f\"Message {i} from producer {producer_id}\"))\n        producer.send_batch(event_data_batch)\n        print(f\"Producer {producer_id} sent batch.\")\n\ncredential = DefaultAzureCredential()\nproducer = EventHubProducerClient(\n    fully_qualified_namespace=EVENTHUB_NAMESPACE,\n    eventhub_name=EVENTHUB_NAME,\n    credential=credential\n)\n\nwith producer:\n    with ThreadPoolExecutor(max_workers=5) as executor:\n        for i in range(5):  # Launch 5 threads\n            executor.submit(send_batch, i, producer)\n```\n\nFor scenarios requiring concurrent sending in asyncio applications, ensure proper coroutine-safety management using mechanisms like asyncio.Lock()\n```python\nimport asyncio\nfrom azure.eventhub.aio import EventHubProducerClient\nfrom azure.eventhub import EventData\nfrom azure.identity.aio import DefaultAzureCredential\n\nEVENTHUB_NAMESPACE = \"<your-namespace>.servicebus.windows.net\"\nEVENTHUB_NAME = \"<your-eventhub-name>\"\n\n# Shared lock for coroutine-safe access\nproducer_lock = asyncio.Lock()\n\nasync def send_batch(producer_id, producer):\n    async with producer_lock:\n        event_data_batch = await producer.create_batch()\n        for i in range(10):\n            event_data_batch.add(EventData(f\"Message {i} from producer {producer_id}\"))\n        await producer.send_batch(event_data_batch)\n        print(f\"Producer {producer_id} sent batch.\")\n\ncredential = DefaultAzureCredential()\nproducer = EventHubProducerClient(\n    fully_qualified_namespace=EVENTHUB_NAMESPACE,\n    eventhub_name=EVENTHUB_NAME,\n    credential=credential\n)\n\nasync with producer:\n    await asyncio.gather(*(send_batch(i, producer) for i in range(5)))\n```\n\n## Examples\n\nThe following sections provide several code snippets covering some of the most common Event Hubs tasks, including:\n\n- [Inspect an Event Hub](#inspect-an-event-hub)\n- [Publish events to an Event Hub](#publish-events-to-an-event-hub)\n- [Consume events from an Event Hub](#consume-events-from-an-event-hub)\n- [Consume events from an Event Hub in batches](#consume-events-from-an-event-hub-in-batches)\n- [Publish events to an Event Hub asynchronously](#publish-events-to-an-event-hub-asynchronously)\n- [Consume events from an Event Hub asynchronously](#consume-events-from-an-event-hub-asynchronously)\n- [Consume events from an Event Hub in batches asynchronously](#consume-events-from-an-event-hub-in-batches-asynchronously)\n- [Consume events and save checkpoints using a checkpoint store](#consume-events-and-save-checkpoints-using-a-checkpoint-store)\n- [Use EventHubConsumerClient to work with IoT Hub](#use-eventhubconsumerclient-to-work-with-iot-hub)\n\n### Inspect an Event Hub\n\nGet the partition ids of an Event Hub.\n\n```python\nimport os\nfrom azure.eventhub import EventHubConsumerClient\nfrom azure.identity import DefaultAzureCredential\n\nFULLY_QUALIFIED_NAMESPACE = os.environ[\"EVENT_HUB_HOSTNAME\"]\nEVENTHUB_NAME = os.environ['EVENT_HUB_NAME']\n\nconsumer_client = EventHubConsumerClient(\n    fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,\n    consumer_group='$Default',\n    eventhub_name=EVENTHUB_NAME,\n    credential=DefaultAzureCredential(),\n)\n\nwith consumer_client:\n    pass # consumer_client is now ready to be used.\n```\n\n### Publish events to an Event Hub\n\nUse the `create_batch` method on `EventHubProducerClient` to create an `EventDataBatch` object which can then be sent using the `send_batch` method.\nEvents may be added to the `EventDataBatch` using the `add` method until the maximum batch size limit in bytes has been reached.\n\n<!-- SNIPPET:send.send_event_data_batch -->\n\n```python\ndef send_event_data_batch(producer):\n    # Without specifying partition_id or partition_key\n    # the events will be distributed to available partitions via round-robin.\n    event_data_batch = producer.create_batch()\n    event_data_batch.add(EventData(\"Single message\"))\n    producer.send_batch(event_data_batch)\n```\n\n<!-- END SNIPPET -->\n\n### Consume events from an Event Hub\n\nThere are multiple ways to consume events from an EventHub.  To simply trigger a callback when an event is received,\nthe `EventHubConsumerClient.receive` method will be of use as follows:\n\n```python\nimport logging\nfrom azure.eventhub import EventHubConsumerClient\nfrom azure.identity import DefaultAzureCredential\n\nfully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'\nconsumer_group = '<< CONSUMER GROUP >>'\neventhub_name = '<< NAME OF THE EVENT HUB >>'\nclient = EventHubConsumerClient(\n    fully_qualified_namespace=fully_qualified_namespace,\n    eventhub_name=eventhub_name,\n    consumer_group=consumer_group,\n    credential=DefaultAzureCredential(),\n)\n\nlogger = logging.getLogger(\"azure.eventhub\")\nlogging.basicConfig(level=logging.INFO)\n\ndef on_event(partition_context, event):\n    logger.info(\"Received event from partition {}\".format(partition_context.partition_id))\n    partition_context.update_checkpoint(event)\n\nwith client:\n    client.receive(\n        on_event=on_event,\n        starting_position=\"-1\",  # \"-1\" is from the beginning of the partition.\n    )\n    # receive events from specified partition:\n    # client.receive(on_event=on_event, partition_id='0')\n```\n\n### Consume events from an Event Hub in batches\n\nWhereas the above sample triggers the callback for each message as it is received, the following sample\ntriggers the callback on a batch of events, attempting to receive a number at a time.\n\n```python\nimport logging\nfrom azure.eventhub import EventHubConsumerClient\nfrom azure.identity import DefaultAzureCredential\n\nfully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'\nconsumer_group = '<< CONSUMER GROUP >>'\neventhub_name = '<< NAME OF THE EVENT HUB >>'\nclient = EventHubConsumerClient(\n    fully_qualified_namespace=fully_qualified_namespace,\n    eventhub_name=eventhub_name,\n    consumer_group=consumer_group,\n    credential=DefaultAzureCredential(),\n)\n\nlogger = logging.getLogger(\"azure.eventhub\")\nlogging.basicConfig(level=logging.INFO)\n\ndef on_event_batch(partition_context, events):\n    logger.info(\"Received event from partition {}\".format(partition_context.partition_id))\n    partition_context.update_checkpoint()\n\nwith client:\n    client.receive_batch(\n        on_event_batch=on_event_batch,\n        starting_position=\"-1\",  # \"-1\" is from the beginning of the partition.\n    )\n    # receive events from specified partition:\n    # client.receive_batch(on_event_batch=on_event_batch, partition_id='0')\n```\n\n### Publish events to an Event Hub asynchronously\n\nUse the `create_batch` method on `EventHubProducer` to create an `EventDataBatch` object which can then be sent using the `send_batch` method.\nEvents may be added to the `EventDataBatch` using the `add` method until the maximum batch size limit in bytes has been reached.\n```python\nimport asyncio\nfrom azure.eventhub.aio import EventHubProducerClient   # The package name suffixed with \".aio\" for async\nfrom azure.eventhub import EventData\nfrom azure.identity.aio import DefaultAzureCredential\n\nfully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'\neventhub_name = '<< NAME OF THE EVENT HUB >>'\n\nasync def create_batch(client):\n    event_data_batch = await client.create_batch()\n    can_add = True\n    while can_add:\n        try:\n            event_data_batch.add(EventData('Message inside EventBatchData'))\n        except ValueError:\n            can_add = False  # EventDataBatch object reaches max_size.\n    return event_data_batch\n\nasync def send():\n    client = EventHubProducerClient(\n        fully_qualified_namespace=fully_qualified_namespace,\n        eventhub_name=eventhub_name,\n        credential=DefaultAzureCredential(),\n    )\n    batch_data = await create_batch(client)\n    async with client:\n        await client.send_batch(batch_data)\n\nif __name__ == '__main__':\n    loop = asyncio.get_event_loop()\n    loop.run_until_complete(send())\n```\n\n### Consume events from an Event Hub asynchronously\n\nThis SDK supports both synchronous and asyncio based code.  To receive as demonstrated in the samples above, but within\naio, one would need the following:\n\n```python\nimport logging\nimport asyncio\nfrom azure.eventhub.aio import EventHubConsumerClient\nfrom azure.identity.aio import DefaultAzureCredential\n\nfully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'\nconsumer_group = '<< CONSUMER GROUP >>'\neventhub_name = '<< NAME OF THE EVENT HUB >>'\n\nlogger = logging.getLogger(\"azure.eventhub\")\nlogging.basicConfig(level=logging.INFO)\n\nasync def on_event(partition_context, event):\n    logger.info(\"Received event from partition {}\".format(partition_context.partition_id))\n    await partition_context.update_checkpoint(event)\n\nasync def receive():\n    client = EventHubConsumerClient(\n        fully_qualified_namespace=fully_qualified_namespace,\n        eventhub_name=eventhub_name,\n        consumer_group=consumer_group,\n        credential=DefaultAzureCredential(),\n    )\n    async with client:\n        await client.receive(\n            on_event=on_event,\n            starting_position=\"-1\",  # \"-1\" is from the beginning of the partition.\n        )\n        # receive events from specified partition:\n        # await client.receive(on_event=on_event, partition_id='0')\n\nif __name__ == '__main__':\n    loop = asyncio.get_event_loop()\n    loop.run_until_complete(receive())\n```\n\n### Consume events from an Event Hub in batches asynchronously\n\nAll synchronous functions are supported in aio as well.  As demonstrated above for synchronous batch receipt, one can accomplish\nthe same within asyncio as follows:\n\n```python\nimport logging\nimport asyncio\nfrom azure.eventhub.aio import EventHubConsumerClient\nfrom azure.identity.aio import DefaultAzureCredential\n\nfully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'\nconsumer_group = '<< CONSUMER GROUP >>'\neventhub_name = '<< NAME OF THE EVENT HUB >>'\n\nlogger = logging.getLogger(\"azure.eventhub\")\nlogging.basicConfig(level=logging.INFO)\n\nasync def on_event_batch(partition_context, events):\n    logger.info(\"Received event from partition {}\".format(partition_context.partition_id))\n    await partition_context.update_checkpoint()\n\nasync def receive_batch():\n    client = EventHubConsumerClient(\n        fully_qualified_namespace=fully_qualified_namespace,\n        eventhub_name=eventhub_name,\n        consumer_group=consumer_group,\n        credential=DefaultAzureCredential(),\n    )\n    async with client:\n        await client.receive_batch(\n            on_event_batch=on_event_batch,\n            starting_position=\"-1\",  # \"-1\" is from the beginning of the partition.\n        )\n        # receive events from specified partition:\n        # await client.receive_batch(on_event_batch=on_event_batch, partition_id='0')\n\nif __name__ == '__main__':\n    loop = asyncio.get_event_loop()\n    loop.run_until_complete(receive_batch())\n```\n\n### Consume events and save checkpoints using a checkpoint store\n\n`EventHubConsumerClient` is a high level construct which allows you to receive events from multiple partitions at once\nand load balance with other consumers using the same Event Hub and consumer group.\n\nThis also allows the user to track progress when events are processed using checkpoints.\n\nA checkpoint is meant to represent the last successfully processed event by the user from a particular partition of\na consumer group in an Event Hub instance. The `EventHubConsumerClient` uses an instance of `CheckpointStore` to update checkpoints\nand to store the relevant information required by the load balancing algorithm.\n\nSearch pypi with the prefix `azure-eventhub-checkpointstore` to\nfind packages that support this and use the `CheckpointStore` implementation from one such package. Please note that both sync and async libraries are provided.\n\nIn the below example, we create an instance of `EventHubConsumerClient` and use a `BlobCheckpointStore`. You need\nto [create an Azure Storage account](https://learn.microsoft.com/azure/storage/common/storage-quickstart-create-account?tabs=azure-portal)\nand a [Blob Container](https://learn.microsoft.com/azure/storage/blobs/storage-quickstart-blobs-portal#create-a-container) to run the code.\n\n[Azure Blob Storage Checkpoint Store Async](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub-checkpointstoreblob-aio)\nand [Azure Blob Storage Checkpoint Store Sync](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub-checkpointstoreblob)\nare one of the `CheckpointStore` implementations we provide that applies Azure Blob Storage as the persistent store.\n\n\n```python\nimport asyncio\n\nfrom azure.eventhub.aio import EventHubConsumerClient\nfrom azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore\nfrom azure.identity.aio import DefaultAzureCredential\n\nfully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'\nconsumer_group = '<< CONSUMER GROUP >>'\neventhub_name = '<< NAME OF THE EVENT HUB >>'\nblob_account_url = '<< STORAGE ACCOUNT URL >>'\ncontainer_name = '<<NAME OF THE BLOB CONTAINER>>'\n\nasync def on_event(partition_context, event):\n    # do something\n    await partition_context.update_checkpoint(event)  # Or update_checkpoint every N events for better performance.\n\nasync def receive(client):\n    await client.receive(\n        on_event=on_event,\n        starting_position=\"-1\",  # \"-1\" is from the beginning of the partition.\n    )\n\nasync def main():\n    checkpoint_store = BlobCheckpointStore(\n        blob_account_url=blob_account_url,\n        container_name=container_name,\n        credential=DefaultAzureCredential()\n    )\n    client = EventHubConsumerClient(\n        fully_qualified_namespace=fully_qualified_namespace,\n        eventhub_name=eventhub_name,\n        credential=DefaultAzureCredential(),\n        consumer_group=consumer_group,\n        checkpoint_store=checkpoint_store,  # For load balancing and checkpoint. Leave None for no load balancing\n    )\n    async with client:\n        await receive(client)\n\nif __name__ == '__main__':\n    loop = asyncio.get_event_loop()\n    loop.run_until_complete(main())\n```\n\n### Use EventHubConsumerClient to work with IoT Hub\n\nYou can use `EventHubConsumerClient` to work with IoT Hub as well. This is useful for receiving telemetry data of IoT Hub from the\nlinked EventHub. The associated connection string will not have send claims, hence sending events is not possible.\n\nPlease notice that the connection string needs to be for an [Event Hub-compatible endpoint](https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-messages-read-builtin),\ne.g. \"Endpoint=sb://my-iothub-namespace-[uid].servicebus.windows.net/;SharedAccessKeyName=my-SA-name;SharedAccessKey=my-SA-key;EntityPath=my-iot-hub-name\"\n\nThere are two ways to get the Event Hubs compatible endpoint:\n- Manually get the \"Built-in endpoints\" of the IoT Hub in Azure Portal and receive from it.\n```python\nfrom azure.eventhub import EventHubConsumerClient\n\nconnection_str = 'Endpoint=sb://my-iothub-namespace-[uid].servicebus.windows.net/;SharedAccessKeyName=my-SA-name;SharedAccessKey=my-SA-key;EntityPath=my-iot-hub-name'\nconsumer_group = '<< CONSUMER GROUP >>'\nclient = EventHubConsumerClient.from_connection_string(connection_str, consumer_group)\n\npartition_ids = client.get_partition_ids()\n```\n- Programmatically retrieve the built-in Event Hubs compatible endpoint.\nRefer to [IoT Hub Connection String Sample](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/samples/async_samples/iot_hub_connection_string_receive_async.py).\n\n## Troubleshooting\n\nSee the `azure-eventhub` [troubleshooting guide](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/TROUBLESHOOTING.md) for details on how to diagnose various failure scenarios.\n\n### Logging\n\n- Enable `azure.eventhub` logger to collect traces from the library.\n- Enable AMQP frame level trace by setting `logging_enable=True` when creating the client.\n- Refer to [this guide](https://learn.microsoft.com/azure/developer/python/sdk/azure-sdk-logging) on configuring logging for Azure libraries for Python for additional information.\n\n```python\nimport logging\nimport sys\n\nhandler = logging.StreamHandler(stream=sys.stdout)\nlog_fmt = logging.Formatter(fmt=\"%(asctime)s | %(threadName)s | %(levelname)s | %(name)s | %(message)s\")\nhandler.setFormatter(log_fmt)\nlogger = logging.getLogger('azure.eventhub')\nlogger.setLevel(logging.DEBUG)\nlogger.addHandler(handler)\n\n...\n\nfrom azure.eventhub import EventHubProducerClient, EventHubConsumerClient\n\nproducer = EventHubProducerClient(..., logging_enable=True)\nconsumer = EventHubConsumerClient(..., logging_enable=True)\n```\n\n## Next steps\n\n### More sample code\n\nPlease take a look at the [samples](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/eventhub/azure-eventhub/samples) directory for detailed examples of how to use this library to send and receive events to/from Event Hubs.\n\n### Documentation\n\nReference documentation is available [here](https://azuresdkdocs.z19.web.core.windows.net/python/azure-eventhub/latest/azure.eventhub.html).\n\n### Schema Registry and Avro Encoder\n\nThe EventHubs SDK integrates nicely with the [Schema Registry][schemaregistry_service] service and [Avro][avro].\nFor more information, please refer to [Schema Registry SDK][schemaregistry_repo] and [Schema Registry Avro Encoder SDK][schemaregistry_avroencoder_repo].\n\n### Pure Python AMQP Transport and Backward Compatibility Support\n\nThe Azure Event Hubs client library is now based on a pure Python AMQP implementation. `uAMQP` has been removed as required dependency.\n\nTo use `uAMQP` as the underlying transport:\n\n1. Install `uamqp` with pip.\n\n```\n$ pip install uamqp \n```\n\n2. Pass `uamqp_transport=True` during client construction.\n\n```python\nfrom azure.eventhub import EventHubProducerClient, EventHubConsumerClient\nfrom azure.identity import DefaultAzureCredential\n\nfully_qualified_namespace = '<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>'\nconsumer_group = '<< CONSUMER GROUP >>'\neventhub_name = '<< NAME OF THE EVENT HUB >>'\n\nclient = EventHubProducerClient(\n    fully_qualified_namespace=fully_qualified_namespace,\n    eventhub_name=eventhub_name,\n    credential=DefaultAzureCredential(),\n    uamqp_transport=True\n)\nclient = EventHubConsumerClient(\n    fully_qualified_namespace=fully_qualified_namespace,\n    eventhub_name=eventhub_name,\n    credential=DefaultAzureCredential(),\n    consumer_group=consumer_group,\n    uamqp_transport=True\n)\n```\n\nNote: The `message` attribute on `EventData`/`EventDataBatch`, which previously exposed the `uamqp.Message`, has been deprecated.\n The \"Legacy\" objects returned by `EventData.message`/`EventDataBatch.message` have been introduced to help facilitate the transition.\n\n### Building uAMQP wheel from source\n\nIf [uAMQP](https://pypi.org/project/uamqp/) is intended to be used as the underlying AMQP protocol implementation for `azure-eventhub`,\nuAMQP wheels can be found for most major operating systems.\n\nIf you intend to use `uAMQP` and you're running on a platform for which uAMQP wheels are not provided, please follow\n the [uAMQP Installation](https://github.com/Azure/azure-uamqp-python#installation) guidance to install from source.\n\n### Provide Feedback\n\nIf you encounter any bugs or have suggestions, please file an issue in the [Issues](https://github.com/Azure/azure-sdk-for-python/issues) section of the project.\n\n## Contributing\n\nThis project welcomes contributions and suggestions.  Most contributions require you to agree to a\nContributor License Agreement (CLA) declaring that you have the right to, and actually do, grant us the rights to use your contribution. For details, visit https://cla.microsoft.com.\n\nWhen you submit a pull request, a CLA-bot will automatically determine whether you need to provide a CLA and decorate the\nPR appropriately (e.g., label, comment). Simply follow the instructions provided by the bot. You will only need to do this once across all repos using our CLA.\n\nThis project has adopted the [Microsoft Open Source Code of Conduct](https://opensource.microsoft.com/codeofconduct/).\nFor more information see the [Code of Conduct FAQ](https://opensource.microsoft.com/codeofconduct/faq/) or contact [opencode@microsoft.com](mailto:opencode@microsoft.com) with any additional questions or comments.\n\n[avro]: https://avro.apache.org/\n[api_reference]: https://learn.microsoft.com/python/api/overview/azure/eventhub-readme\n[schemaregistry_service]: https://aka.ms/schemaregistry\n[schemaregistry_repo]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/schemaregistry/azure-schemaregistry\n[schemaregistry_avroencoder_repo]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/schemaregistry/azure-schemaregistry-avroencoder\n\n\n",
    "bugtrack_url": null,
    "license": null,
    "summary": "Microsoft Azure Event Hubs Client Library for Python",
    "version": "5.15.1",
    "project_urls": {
        "repository": "https://github.com/Azure/azure-sdk-for-python/tree/main/sdk"
    },
    "split_keywords": [
        "azure",
        " azure sdk"
    ],
    "urls": [
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "30bab54e0f76384b00a8682b1e041673f43947f668dbc8508f026d67446a7d2f",
                "md5": "6e01a338cb846921428574a17cab17cb",
                "sha256": "ebe1379d228632d12892f0e2d480e0c13a234c384249b0c2c0fe862cd58b3e48"
            },
            "downloads": -1,
            "filename": "azure_eventhub-5.15.1-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "6e01a338cb846921428574a17cab17cb",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.9",
            "size": 317133,
            "upload_time": "2025-10-30T11:28:23",
            "upload_time_iso_8601": "2025-10-30T11:28:23.395836Z",
            "url": "https://files.pythonhosted.org/packages/30/ba/b54e0f76384b00a8682b1e041673f43947f668dbc8508f026d67446a7d2f/azure_eventhub-5.15.1-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "20bf96eec5aa57cb08f1e84e1dfa3888e52dbba53e218596cdab5b61d795a1ae",
                "md5": "32c15c27e2e5e628d614f2de8bba5360",
                "sha256": "201de7b6e10b7517c594e529aa7d21c6887ac2a4757fc4850c991a3f640a02fb"
            },
            "downloads": -1,
            "filename": "azure_eventhub-5.15.1.tar.gz",
            "has_sig": false,
            "md5_digest": "32c15c27e2e5e628d614f2de8bba5360",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.9",
            "size": 378879,
            "upload_time": "2025-10-30T11:28:21",
            "upload_time_iso_8601": "2025-10-30T11:28:21.740029Z",
            "url": "https://files.pythonhosted.org/packages/20/bf/96eec5aa57cb08f1e84e1dfa3888e52dbba53e218596cdab5b61d795a1ae/azure_eventhub-5.15.1.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-10-30 11:28:21",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "github_user": "Azure",
    "github_project": "azure-sdk-for-python",
    "travis_ci": false,
    "coveralls": true,
    "github_actions": true,
    "lcname": "azure-eventhub"
}