Microsoft 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?
The Azure Event Hubs client library allows for publishing and consuming of Azure Event Hubs events and may be used to:
Source code | Package (PyPi) | Package (Conda) | API reference documentation | Product documentation | Samples
Python 3.7 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.
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. 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 Azure Event Hubs client library for Python with pip:
$ pip install azure-eventhub
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:
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.
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.Create client using the azure-identity library:
Alternately, one can use a Credential object to authenticate via AAD with the azure-identity package.
TokenCredential
protocol available in the
azure-identity package. The host name is of the format <yournamespace.servicebus.windows.net>
.azure-identity
, please install the package:
pip install azure-identity
aiohttp
:
pip install aiohttp
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. Also, the concepts for AMQP are well documented in OASIS Advanced Messaging Queuing Protocol (AMQP) Version 1.0.
We do not guarantee that the EventHubProducerClient or EventHubConsumerClient are thread-safe. We do not recommend reusing these instances across threads. It is up to the running application to use these classes in a thread-safe manner.
The data model type, EventDataBatch
is not thread-safe. It should not be shared across threads nor used concurrently with client methods.
The following sections provide several code snippets covering some of the most common Event Hubs tasks, including:
Get the partition ids of an Event Hub.
import os
from azure.eventhub import EventHubConsumerClient
CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"]
EVENTHUB_NAME = os.environ['EVENT_HUB_NAME']
consumer_client = EventHubConsumerClient.from_connection_string(
conn_str=CONNECTION_STR,
consumer_group='$Default',
eventhub_name=EVENTHUB_NAME,
)
with consumer_client:
pass # consumer_client is now ready to be used.
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.
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)
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:
import logging
from azure.eventhub import EventHubConsumerClient
connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)
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')
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.
import logging
from azure.eventhub import EventHubConsumerClient
connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)
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')
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.
import asyncio
from azure.eventhub.aio import EventHubProducerClient # The package name suffixed with ".aio" for async
from azure.eventhub import EventData
connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
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.from_connection_string(connection_str, eventhub_name=eventhub_name)
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())
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:
import logging
import asyncio
from azure.eventhub.aio import EventHubConsumerClient
connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS 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.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)
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())
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:
import logging
import asyncio
from azure.eventhub.aio import EventHubConsumerClient
connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS 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.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)
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())
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
and a Blob Container to run the code.
Azure Blob Storage Checkpoint Store Async
and Azure Blob Storage Checkpoint Store Sync
are one of the CheckpointStore
implementations we provide that applies Azure Blob Storage as the persistent store.
import asyncio
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
storage_connection_str = '<< CONNECTION STRING FOR THE STORAGE >>'
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.from_connection_string(storage_connection_str, container_name)
client = EventHubConsumerClient.from_connection_string(
connection_str,
consumer_group,
eventhub_name=eventhub_name,
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())
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, 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:
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()
See the azure-eventhubs
troubleshooting guide for details on how to diagnose various failure scenarios.
Please take a look at the samples directory for detailed examples of how to use this library to send and receive events to/from Event Hubs.
Reference documentation is available here.
The EventHubs SDK integrates nicely with the Schema Registry service and Avro. For more information, please refer to Schema Registry SDK and Schema Registry Avro Encoder SDK.
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:
uamqp
with pip.$ pip install uamqp
uamqp_transport=True
during client construction.from azure.eventhub import EventHubProducerClient, EventHubConsumerClient
connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
client = EventHubProducerClient.from_connection_string(
connection_str, eventhub_name=eventhub_name, uamqp_transport=True
)
client = EventHubConsumerClient.from_connection_string(
connection_str, consumer_group, eventhub_name=eventhub_name, 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.
If 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 guidance to install from source.
If you encounter any bugs or have suggestions, please file an issue in the Issues section of the project.
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. For more information see the Code of Conduct FAQ or contact opencode@microsoft.com with any additional questions or comments.
socket_timeout
has been added to the sync and async EventHubConsumerClient
and EventHubProducerClient
.BlobCheckpointStore
was reprocessing old events after an error.end frame received on invalid channel
which was raised when a disconnect was sent by the service (#30860)link already closed
which was raised when the client was closing and disconnecting from the service (#30836)Azure.EventHubs.send
to EventHubs.send
Azure.EventHubs.message
to EventHubs.message
Azure.EventHubs.process
to EventHubs.process
EventHubs.receive
span will be created upon receiving events.messaging.system
- messaging system (i.e., eventhubs
)messaging.operation
- type of operation (i.e., publish
, receive
, or process
)messaging.batch.message_count
- number of messages sent, received, or processed (if more than one)component
attribute was removed from all spans.send
spans now contain links to message
spans. Now, message
spans will no longer contain a link to the send
span.traceparent
(and tracestate
if applicable)websocket-client
was not installed, the error was not caught/raised properly (issue #28453).Version 5.11.0 is our first stable release of the Azure Event Hubs client library based on a pure Python implemented AMQP stack.
uamqp_transport
has been added to sync and async EventHubProducerClient
/EventHubConsumerClient
constructors which indicates whether to use the uamqp
library or the default pure Python AMQP library as the underlying transport.EventHubSharedKeyCredential
returned an AccessToken.token
of type bytes
and not str
, now matching the documentation.message
attribute on EventData
/EventDataBatch
, which previously exposed the uamqp.Message
, has been deprecated.
LegacyMessage
/LegacyBatchMessage
objects returned by the message
attribute on EventData
/EventDataBatch
have been introduced to help facilitate the transition.uamqp >= 1.6.3
as an optional dependency for use with the uamqp_transport
keyword.
websocket-client
to aiohttp
(Issue #24315, thanks @hansmbakker for the suggestion).This version and all future versions will require Python 3.7+. Python 3.6 is no longer supported.
uamqp_transport
optional parameter to the clients, to allow switching to the uamqp
library as the transport.This version and all future versions will require Python 3.7+, Python 3.6 is no longer supported.
BufferedProducer
that would block when flushing the queue causing the client to freeze up (issue #23510).EventHubProducerClient
and EventHubConsumerClient
that set the default value of the transport_type
parameter in the from_connection_string
methods to None
rather than TransportType.Amqp
.send_event
to EventHubProducerClient
which allows sending single EventData
or AmqpAnnotatedMessage
.EventHubProducerClient
which is intended to allow for efficient publishing of events
without having to explicitly manage batches in the application.
EventHubProducerClient
and from_connection_string
method takes the following new keyword arguments
for configuration:
buffered_mode
: The flag to enable/disable buffered mode sending.on_success
: The callback to be called once events have been successfully published.on_error
: The callback to be called once events have failed to be published.max_buffer_length
: The total number of events per partition that can be buffered before a flush will be triggered.max_wait_time
: The amount of time to wait for a batch to be built with events in the buffer before publishing.EventHubProducerClient.flush
which flushes events in the buffer to be sent immediately.EventHubProducerClient.get_buffered_event_count
which returns the number of events that are buffered and waiting to be published for a given partition.EventHubProducerClient.total_buffered_event_count
which returns the total number of events that are currently buffered and waiting to be published, across all partitions.flush
to EventHubProducerClient.close
which indicates whether to flush the buffer or not while closing.from_message_content
has been added to EventData
for interoperability with the Schema Registry Avro Encoder library, and takes content
and content_type
as positional parameters.send_event
to EventHubProducerClient
which allows sending single EventData
or AmqpAnnotatedMessage
.EventHubProducerClient
which is intended to allow for efficient publishing of events
without having to explicitly manage batches in the application.
EventHubProducerClient
and from_connection_string
method now takes the following new keyword arguments
for configuration:
buffered_mode
: The flag to enable/disable buffered mode sending.on_success
: The callback to be called once events have been successfully published.on_error
: The callback to be called once events have failed to be published.max_buffer_length
: The total number of events per partition that can be buffered before a flush will be triggered.max_wait_time
: The amount of time to wait for a batch to be built with events in the buffer before publishing.EventHubProducerClient.flush
which flushes events in the buffer to be sent immediately.EventHubProducerClient.get_buffered_event_count
which returns the number of events that are buffered and waiting to be published for a given partition.EventHubProducerClient.total_buffered_event_count
which returns the total number of events that are currently buffered and waiting to be published, across all partitions.flush
to EventHubProducerClient.close
which indicates whether to flush the buffer or not while closing.EventData
internals for interoperability with the Schema Registry Avro Encoder library.from_message_data
on EventData
has been renamed from_message_content
for interoperability with the Schema Registry Avro Encoder library. The data
parameter has been renamed to content
.EventHubProducerClient
and EventHubConsumerClient
which will be added back in future previews as we work towards a stable release:
from_connection_string
methods of the EventHubProducerClient
and EventHubConsumerClient
is not supported: transport_type
, http_proxy
, custom_endpoint_address
, and connection_verify
.EventHubProducerClient
and EventHubConsumerClient
.Version 5.8.0a1 is our first efforts to build an Azure Event Hubs client library based on pure python implemented AMQP stack.
from_connection_string
methods of the EventHubProducerClient
and EventHubConsumerClient
is not supported: transport_type
, http_proxy
, custom_endpoint_address
, and connection_verify
.This version and all future versions will require Python 3.6+. Python 2.7 is no longer supported.
EventHubProducerClient
and EventHubConsumerClient
constructors and from_connection_string
take retry_mode
as a keyword argument.EventHubProducerClient
could be reopened for sending events instead of encountering with KeyError
when the client is previously closed (issue #21849).azure.eventhub.amqp.AmqpMessageHeader
and azure.eventhub.amqp.AmqpMessageProperties
contain specific properties using the in
keyword.azure.eventhub.amqp
.azure.eventhub.amqp.AmqpMessageBodyType
to represent the body type of the message which includes:
DATA
: The body of message consists of one or more data sections and each section contains opaque binary data.SEQUENCE
: The body of message consists of one or more sequence sections and each section contains an arbitrary number of structured data elements.VALUE
: The body of message consists of one amqp-value section and the section contains a single AMQP value.azure.eventhub.amqp.AmqpAnnotatedMessage
for accessing low-level amqp message sections which can be instantiated for sending.azure.eventhub.amqp.AmqpMessageHeader
and azure.eventhub.amqp.AmqpMessageProperties
for accessing amqp header and properties.body_type
on azure.eventhub.EventData
which returns azure.eventhub.amqp.AmqpMessageBodyType
.raw_amqp_message
on azure.eventhub.EventData
which returns azure.eventhub.amqp.AmqpAnnotatedMessage
.New Features
azure.core.credentials.AzureNamedKeyCredential
as credential for authenticating producer and consumer clients.Bug Fixes
Notes
This version follows from version 5.3.1, rather than 5.4.0b1 so that the preview idempotent producer feature is not included.
New Features
azure.core.credentials.AzureSasCredential
as credential for authenticating producer and consumer clients.list_ownership
, claim_ownership
, update_checkpoint
, list_checkpoints
on sync and async CheckpointStore
to support taking **kwargs
.
**kwargs
in the methods listed previously will result in the following pylint error: W0221: Parameters differ from overridden ________ method (arguments-differ)
.update_checkpoint
on sync and async PartitionContext
to support taking **kwargs
.Bug Fixes
Notes
This version and all future versions will require Python 2.7 or Python 3.6+, Python 3.5 is no longer supported.
New Features
EventHubProducerClient
constructor accepts two new parameters for idempotent publishing:
enable_idempotent_partitions
: A boolean value to tell the EventHubProducerClient
whether to enable idempotency.partition_config
: The set of configurations that can be specified to influence publishing behavior
specific to the configured Event Hub partition.get_partition_publishing_properties
on EventHubProducerClient
to inspect the information
about the state of publishing for a partition.published_sequence_number
on EventData
to get the publishing sequence number assigned
to the event at the time it was successfully published.starting_published_sequence_number
on EventDataBatch
to get the publishing sequence
number assigned to the first event in the batch at the time the batch was successfully published.azure.eventhub.PartitionPublishingConfiguration
which is a set of configurations that can be
specified to influence the behavior when publishing directly to an Event Hub partition.Notes
This version will be the last version to officially support Python 3.5, future versions will require Python 2.7 or Python 3.6+.
Bug fixes
event_data_batch
will be a no-op now instead of raising error.New Features
parse_connection_string
method which parses a connection string into a properties bag, EventHubConnectionStringProperties
, containing its component parts.from_connection_string
method of EventHubConsumerClient
and EventHubProducerClient
now accept two new optional arguments:
custom_endpoint_address
which allows for specifying a custom endpoint to use when communicating with the Event Hubs service,
and is useful when your network does not allow communicating to the standard Event Hubs endpoint.connection_verify
which allows for specifying the path to the custom CA_BUNDLE file of the SSL certificate which is used to authenticate
the identity of the connection endpoint.Notes
Bug fixes
azure.eventhub.extension.__init__.py
to be compatible with pkgutil-style namespace (PR #13210, thanks @pjachowi).uamqp.ReceiveClient
and uamqp.ReceiveClientAsync
receive messages during connection establishment (#15555).New Features
from_connection_string
methods now supports using the SharedAccessSignature
key in leiu of sharedaccesskey
and sharedaccesskeyname
, taking the string of the properly constructed token as value.New Features
EventHubConsumerClient
constructor accepts two new parameters for the load balancer.
load_balancing_strategy
, which can be "greedy" or "balanced".
With greedy strategy, one execution of load balancing will claim as many partitions as required to balance the load
whereas with balanced strategy one execution of load balancing will claim at most 1 partition.partition_ownership_expiration_interval
, which allows you to customize the partition ownership expiration for load balancing.
A consumer client may lose its owned partitions more often with a smaller expiration interval. But a larger interval
may result in idle partitions not being claimed for longer time.azure.eventhub.LoadBalancingStrategy
for load_balancing_strategy
.New Features
EventHubProducerClient.send_batch
accepts either an EventDataBatch
or a finite list of EventData
. #9181Bug fixes
azure.eventhub.EventhubConsumerClient
into an exclusive receiver when it has no checkpoint store. #11181logging_enable
to True
in EventHubConsumerClient
and EventHubProducerClient
.New Features
EventHubConsumerClient.receive_batch()
to receive and process events in batches instead of one by one. #9184EventHubConsumerCliuent.receive()
has a new param max_wait_time
.
on_event
is called every max_wait_time
when no events are received and max_wait_time
is not None
or 0.PartitionContext.update_checkpoint
is now optional. The last received event is used when param event is not passed in.EventData.system_properties
has added missing properties when consuming messages from IotHub. #10408Bug fixes
azure.eventhub.EventHubConsumerClient
#9660get_eventhub_properties
, get_partition_ids
, and get_partition_properties
to raise
an error on Azure Stack #9920Breaking changes
EventData
application_properties
and deprecated method encode_message()
.EventHubConsumerClient
on_error
would be called when EventHubConsumerClient
failed to claim ownership of partitions.on_partition_close
and on_partition_initialize
would be called in the case of exceptions raised by on_event
callback.
EventHubConsumerClient
would close and re-open the internal partition receiver in this case.EventHubConsumerClient
should resume receiving after recovering from an error has been re-prioritized.
starting_position
is provided, it will resume from starting_posititon
.starting_position
, it will resume from the latest position.PartitionContext
update_checkpoint
would do in-memory checkpoint instead of doing nothing when checkpoint store is not explicitly provided.
EventHubConsumerClient
receiving recovering.get_partition_ids
, get_partition_properties
, get_eventhub_properties
would raise error in the case of service returning an error status code.
AuthenticationError
would be raised when service returning error code 401.ConnectError
would be raised when service returning error code 404.EventHubError
would be raised when service returning other error codes.Breaking changes
azure.eventhub.exceptions
.EventHubSharedKeyCredential
objects for synchronous and asynchronous operations.
For async, import the credentials object from the azure.eventhub.aio
namespace.EventData
application_properties
to properties
.EventData
no longer has attribute last_enqueued_event_properties
- use this on PartitionContext
instead.EvenDataBatch
EventDataBatch.try_add
has been renamed to EventDataBatch.add
.size
to size_in_bytes
.max_size
to max_size_in_bytes
.EventHubConsumerClient
and EventHubProducerClient
get_properties
to get_eventhub_properties
.host
to fully_qualified_namespace
, event_hub_path
to eventhub_name
.get_partition_properties
: partition
to partition_id
.consumer_group_name
to consumer_group
and moved that parameter from receive
method to the constructor of EventHubConsumerClient
.initial_event_position
to starting_position
on the receive
method of EventHubConsumerClient
.event_hub_path
to eventhub_name
in constructor and from_connection_string
method of the client object.EventHubProducerClient.send
has been renamed to send_batch
which will only accept EventDataBatch
object as input.EventHubProducerClient.create_batch
now also takes the partition_id
and partition_key
as optional parameters (which are no longer specified at send).PartitionManager
to CheckpointStore
.on_event
and now operates on a single event rather than a list of events.EventPostition
.
starting_position
parameter of the receive
method accepts offset(str
), sequence number(int
), datetime (datetime.datetime
) or dict
of these types.starting_position_inclusive
parameter of the receive
method accepts bool
or dict
indicating whether the given event position is inclusive or not.PartitionContext
no longer has attribute owner_id
.PartitionContext
now has attribute last_enqueued_event_properties
which is populated if track_last_enqueued_event_properties
is set to True
in the receive
method.New features
idle_timeout
in construct and from_connection_string
to EventHubConsumerClient
and EventHubProducerClient
after which the underlying connection will close if there is no further activity.Breaking changes
EventHubClient
, EventHubConsumer
and EventHubProducer
has been removed. Use EventHubProducerClient
and EventHubConsumerClient
instead.
EventHubProducerClient
as substitution forEventHubProducer
.
EventHubProducerClient
supports sending events to different partitions.EventHubConsumerClient
as substitution for EventHubConsumer
.
EventHubConsumerClient
supports receiving events from single/all partitions.EventData
, all receiving is done via callback method: on_events
.EventHubConsumerClient
has taken on the responsibility of EventProcessor
.
EventHubConsumerClient
now accepts PartitionManager
to do load-balancing and checkpoint.PartitionProcessor
by four independent callback methods accepted by the receive
method on EventHubConsumerClient
.
on_events(partition_context, events)
called when events are received.on_error(partition_context, exception
called when errors occur.on_partition_initialize(partition_context)
called when a partition consumer is opened.on_partition_close(partition_context, reason)
called when a partition consumer is closed.azure.eventhub.common
has been removed. Import from azure.eventhub
instead.azure.eventhub.client_abstract
has been removed. Use azure.eventhub.EventHubProducerClient
or azure.eventhub.EventHubConsumerClient
instead.azure.eventhub.client
has been removed. Use azure.eventhub.EventHubProducerClient
or azure.eventhub.EventHubConsumerClient
instead.azure.eventhub.producer
has been removed. Use azure.eventhub.EventHubProducerClient
instead.azure.eventhub.consumer
has been removed. Use azure.eventhub.EventHubConsumerClient
instead.azure.eventhub.aio.client_async
has been removed. Use azure.eventhub.aio.EventHubProducerClient
or azure.eventhub.aio.EventHubConsumerClient
instead.azure.eventhub.aio.producer_async
has been removed. Use azure.eventhub.aio.EventHubProducerClient
instead.azure.eventhub.aio.consumer_async
has been removed. Use azure.eventhub.aio.EventHubConsumerClient
instead.azure.eventhub.aio.event_processor.event_processor
has been removed. Use azure.eventhub.aio.EventHubConsumerClient
instead.azure.eventhub.aio.event_processor.partition_processor
has been removed. Use callback methods instead.azure.eventhub.aio.event_processor.partition_manager
has been removed. Import from azure.eventhub.aio
instead.azure.eventhub.aio.event_processor.partition_context
has been removed. Import from azure.eventhub.aio
instead.azure.eventhub.aio.event_processor.sample_partition_manager
has been removed.Bug fixes
New features
EventHubConsumer
.
track_last_enqueued_event_properties
in method EventHubClient.create_consumer()
.last_enqueued_event_properties
of EventHubConsumer
which contains sequence_number, offset, enqueued_time and retrieval_time information.Breaking changes
EventHubClient
and read properties or events from an IoT Hub.exception
in method close()
of EventHubConsumer
and EventHubProcuer
.New features
EventProcessor
.BlobPartitionManager
which implements PartitionManager
.
EventProcessor
.EventProcessor
.system_properties
on EventData
.Breaking changes
PartitionProcessor
. For initialization please implement the method initialize
.CheckpointManager
by PartitionContext
.
PartitionContext
has partition context information and method update_checkpoint
.PartitionProcessor
to include PartitionContext
as part of the arguments.EventHub/EventHubConsumer/EventHubProducer
to be private.azure.eventhub.eventprocessor
under aio
package, which now becomes azure.eventhub.aio.eventprocessor
.New features
create_batch
on the EventHubProducer
to create an EventDataBatch
that can then be used to add events until the maximum size is reached.
send()
method to send all the added events to Event Hubs.retry_total
: The total number of attempts to redo the failed operation.backoff_factor
: The delay time factor.backoff_max
: The maximum delay time in total.EventHubClient
.OperationTimeoutError
for send operation.EventProcessor
which replaces the older concept of Event Processor Host. This early preview is intended to allow users to test the new design using a single instance of EventProcessor
. The ability to checkpoints to a durable store will be added in future updates.
EventProcessor
: EventProcessor creates and runs consumers for all partitions of the eventhub.PartitionManager
: PartitionManager defines the interface for getting/claiming ownerships of partitions and updating checkpoints.PartitionProcessor
: PartitionProcessor defines the interface for processing events.CheckpointManager
: CheckpointManager takes responsibility for updating checkpoints during events processing.Breaking changes
EventProcessorHost
was replaced by EventProcessor
, please read the new features for details.max_retries
configuration parameter of the EventHubClient with retry_total
.Version 5.0.0b1 is a preview of our efforts to create a client library that is user friendly and idiomatic to the Python ecosystem. The reasons for most of the changes in this update can be found in the Azure SDK Design Guidelines for Python. For more information, please visit https://aka.ms/azure-sdk-preview1-python.
New features
credential
: The credential object used for authentication which implements TokenCredential
interface of getting tokens.transport_type
: The type of transport protocol that will be used for communicating with the Event Hubs service.max_retries
: The max number of attempts to redo the failed operation when an error happened.get_partition_properties
and get_partition_ids
to EventHubClient.Breaking changes
azure.error.EventHubError
azure.error.ConnectionLostError
azure.error.ConnectError
azure.error.AuthenticationError
azure.error.EventDataError
azure.error.EventDataSendError
add_sender
to create_producer
and add_receiver
to create_consumer
in EventHubClient.get_eventhub_info
to get_properties
of EventHubClient.BugFixes
BugFixes
Features
Changes to AzureStorageCheckpointLeaseManager
parameters to support other connection options (issue #61):
storage_account_name
, storage_account_key
and lease_container_name
arguments are now optional keyword arguments.sas_token
argument that must be specified with storage_account_name
in place of storage_account_key
.endpoint_suffix
argument to support storage endpoints in National Clouds.connection_string
argument that, if specified, overrides all other endpoint arguments.lease_container_name
argument now defaults to "eph-leases"
if not specified.Fix for clients failing to start if run called multipled times (issue #64).
Added convenience methods body_as_str
and body_as_json
to EventData object for easier processing of message data.
_async
module to async_ops
for docs generation.auth_timeout
parameter to EventHubClient
and EventHubClientAsync
to configure how long to allow for token
negotiation to complete. Default is 60 seconds.send_timeout
parameter to EventHubClient.add_sender
and EventHubClientAsync.add_async_sender
to determine the
timeout for Events to be successfully sent. Default value is 60 seconds.Stability improvements for EPH.
Updated uAMQP version.
Added new configuration options for Sender and Receiver; keep_alive
and auto_reconnect
.
These flags have been added to the following:
EventHubClient.add_receiver
EventHubClient.add_sender
EventHubClientAsync.add_async_receiver
EventHubClientAsync.add_async_sender
EPHOptions.keey_alive_interval
EPHOptions.auto_reconnect_on_error
EventData.offset
will now return an object of type ~uamqp.common.Offset
rather than str.
The original string value can be retrieved from ~uamqp.common.Offset.value
.async
has been renamed and all classes from
this module can now be imported from azure.eventhub directly.callback
argument from Receiver.receive
and AsyncReceiver.receive
.EventData.properties
has been renamed to EventData.application_properties
.
This removes the potential for messages to be processed via callback for not yet returned
in the batch.namespace_suffix
to EventHubConfig() to support national clouds.device_id
attribute to EventData to support IoT Hub use cases.