Microsoft Azure Event Hubs Client Library for Python
Project description
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:
- 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 | Package (PyPi) | API reference documentation | Product documentation | Samples
Disclaimer
Azure SDK Python packages support for Python 2.7 has ended 01 January 2022. For more information and questions, please refer to https://github.com/Azure/azure-sdk-for-python/issues/20691
This version is our first efforts to build an Azure Event Hubs client library based on pure python implemented AMQP stack. Features like async support or AmqpOverWebsocket are unavailable for this version which will be added back in future previews as we work towards a stable release. Please refer to the changelog for more details.
Getting started
Prerequisites
-
Python 3.6 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 package
Install the Azure Event Hubs client library for Python with pip:
$ pip install azure-eventhub --pre
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:
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.
- The
from_connection_string
method takes the connection string of the formEndpoint=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.
- 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
protocol. There are implementations of the
TokenCredential
protocol available in the azure-identity package. 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
- 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.
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. Also, the concepts for AMQP are well documented in OASIS Advanced Messaging Queuing Protocol (AMQP) Version 1.0.
Examples
The following sections provide several code snippets covering some of the most common Event Hubs tasks, including:
- Inspect an Event Hub
- Publish events to an Event Hub
- Consume events from an Event Hub
- Consume events from an Event Hub in batches
- Consume events and save checkpoints using a checkpoint store
- Use EventHubConsumerClient to work with IoT Hub
Inspect an Event Hub
Get the partition ids of an Event Hub.
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)
partition_ids = client.get_partition_ids()
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.
from azure.eventhub import EventHubProducerClient, EventData
connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'
client = EventHubProducerClient.from_connection_string(connection_str, eventhub_name=eventhub_name)
event_data_batch = 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.
with client:
client.send_batch(event_data_batch)
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:
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')
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.
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')
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.
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 Sync
is one of the CheckpointStore
implementations we provide that applies Azure Blob Storage as the persistent store.
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob 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>>'
def on_event(partition_context, event):
# do something
partition_context.update_checkpoint(event) # Or update_checkpoint every N events for better performance.
def receive(client):
client.receive(
on_event=on_event,
starting_position="-1", # "-1" is from the beginning of the partition.
)
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
)
with client:
receive(client)
if __name__ == '__main__':
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, 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.
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()
Troubleshooting
General
The Event Hubs APIs generate the following exceptions in azure.eventhub.exceptions
- AuthenticationError: Failed to authenticate because of wrong address, SAS policy/key pair, SAS token or azure identity.
- ConnectError: Failed to connect to the EventHubs. The AuthenticationError is a type of ConnectError.
- ConnectionLostError: Lose connection after a connection has been built.
- EventDataError: The EventData to be sent fails data validation. For instance, this error is raised if you try to send an EventData that is already sent.
- EventDataSendError: The Eventhubs service responds with an error when an EventData is sent.
- OperationTimeoutError: EventHubConsumer.send() times out.
- EventHubError: All other Eventhubs related errors. It is also the root error class of all the errors described above.
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.
Next steps
More sample code
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.
Documentation
Reference documentation is available here.
Schema Registry and Avro Serializer
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 Serializer SDK.
Provide Feedback
If you encounter any bugs or have suggestions, please file an issue in the 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. For more information see the Code of Conduct FAQ or contact opencode@microsoft.com with any additional questions or comments.
Release History
5.8.0a1 (2022-01-13)
Version 5.8.0a1 is our first efforts to build an Azure Event Hubs client library based on pure python implemented AMQP stack.
Breaking changes
- The following features have been temporarily pulled out which will be added back in future previews as we work towards a stable release:
- Async is not supported.
- Passing the following keyword arguments to the constructors and
from_connection_string
methods of theEventHubProducerClient
andEventHubConsumerClient
is not supported:transport_type
,http_proxy
,custom_endpoint_address
, andconnection_verify
.
Other Changes
- uAMQP dependency is removed.
5.7.0 (2022-01-12)
This version and all future versions will require Python 3.6+. Python 2.7 is no longer supported.
Features Added
- Added support for fixed (linear) retry backoff:
- Sync/async
EventHubProducerClient
andEventHubConsumerClient
constructors andfrom_connection_string
takeretry_mode
as a keyword argument.
- Sync/async
Bugs Fixed
- Fixed a bug that
EventHubProducerClient
could be reopened for sending events instead of encountering withKeyError
when the client is previously closed (issue #21849).
Other Changes
- Improved token refresh timing to prevent potentially blocking main flow when the token is about to get expired soon.
- Updated uAMQP dependency to 1.5.1.
5.6.1 (2021-10-06)
Bugs Fixed
- Fixed a bug for checking that
azure.eventhub.amqp.AmqpMessageHeader
andazure.eventhub.amqp.AmqpMessageProperties
contain specific properties using thein
keyword.
Other Changes
- Updated uAMQP dependency to 1.4.3.
- Added support for Python 3.10.
- Fixed memory leak in win32 socketio and tlsio (issue #19777).
- Fixed memory leak in the process of converting AMQPValue into string (issue #19777).
5.6.0 (2021-07-07)
Features Added
- Added support for sending AMQP annotated message which allows full access to the AMQP message fields.
- Introduced new namespace
azure.eventhub.amqp
. - Added new enum class
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.
- Introduced new class
azure.eventhub.amqp.AmqpAnnotatedMessage
for accessing low-level amqp message sections which can be instantiated for sending. - Introduced new classes
azure.eventhub.amqp.AmqpMessageHeader
andazure.eventhub.amqp.AmqpMessageProperties
for accessing amqp header and properties. - Added new property
body_type
onazure.eventhub.EventData
which returnsazure.eventhub.amqp.AmqpMessageBodyType
. - Added new read-only property
raw_amqp_message
onazure.eventhub.EventData
which returnsazure.eventhub.amqp.AmqpAnnotatedMessage
.
- Introduced new namespace
Fixed
- Updated uAMQP dependency to 1.4.1.
- Fixed a bug that attributes creation_time, absolute_expiry_time and group_sequence on MessageProperties should be compatible with integer types on Python 2.7.
5.5.0 (2021-05-13)
New Features
- Added support for using
azure.core.credentials.AzureNamedKeyCredential
as credential for authenticating producer and consumer clients.
Bug Fixes
- Fixed bug that custom user agent string should be put in front of the built-in user agent string instead of being appended.
- Updated uAMQP dependency to 1.4.0.
- Fixed memory leaks in the process of link attach where source and target cython objects are not properly deallocated (#15747).
- Improved management operation callback not to parse description value of non AMQP_TYPE_STRING type as string (#18361).
Notes
- Updated azure-core dependency to 1.14.0.
5.4.0 (2021-04-07)
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
- Added support for using
azure.core.credentials.AzureSasCredential
as credential for authenticating producer and consumer clients. - Updated
list_ownership
,claim_ownership
,update_checkpoint
,list_checkpoints
on sync and asyncCheckpointStore
to support taking**kwargs
.- WARNING: Implementing a custom checkpointstore that does not support taking
**kwargs
in the methods listed previously will result in the following pylint error:W0221: Parameters differ from overridden ________ method (arguments-differ)
.
- WARNING: Implementing a custom checkpointstore that does not support taking
- Updated
update_checkpoint
on sync and asyncPartitionContext
to support taking**kwargs
.
Bug Fixes
- Updated uAMQP dependency to 1.3.0.
- Fixed bug that sending message of large size triggering segmentation fault when the underlying socket connection is lost (#13739, #14543).
- Fixed bug in link flow control where link credit and delivery count should be calculated based on per message instead of per transfer frame (#16934).
Notes
- Updated azure-core dependency to 1.13.0.
5.4.0b1 (2021-03-09)
This version and all future versions will require Python 2.7 or Python 3.6+, Python 3.5 is no longer supported.
New Features
- Added support for idempotent publishing which is supported by the service to endeavor to reduce the number of duplicate
events that are published.
EventHubProducerClient
constructor accepts two new parameters for idempotent publishing:enable_idempotent_partitions
: A boolean value to tell theEventHubProducerClient
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.
- Introduced a new method
get_partition_publishing_properties
onEventHubProducerClient
to inspect the information about the state of publishing for a partition. - Introduced a new property
published_sequence_number
onEventData
to get the publishing sequence number assigned to the event at the time it was successfully published. - Introduced a new property
starting_published_sequence_number
onEventDataBatch
to get the publishing sequence number assigned to the first event in the batch at the time the batch was successfully published. - Introduced a new class
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
- Updated uAMQP dependency to 1.2.15.
5.3.1 (2021-03-09)
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
- Sending empty
event_data_batch
will be a no-op now instead of raising error.
5.3.0 (2021-02-08)
New Features
- Added a
parse_connection_string
method which parses a connection string into a properties bag,EventHubConnectionStringProperties
, containing its component parts. - The constructor and
from_connection_string
method ofEventHubConsumerClient
andEventHubProducerClient
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
- Updated uAMQP dependency to 1.2.14.
5.2.1 (2021-01-11)
Bug fixes
- Updated
azure.eventhub.extension.__init__.py
to be compatible with pkgutil-style namespace (PR #13210, thanks @pjachowi). - Updated uAMQP dependency to 1.2.13
- Added support for Python 3.9.
- Fixed bug that macOS was unable to detect network error (#15473).
- Fixed bug that
uamqp.ReceiveClient
anduamqp.ReceiveClientAsync
receive messages during connection establishment (#15555). - Fixed bug where connection establishment on macOS with Clang 12 triggering unrecognized selector exception (#15567).
- Fixed bug in accessing message properties triggering segmentation fault when the underlying C bytes are NULL (#15568).
5.2.0 (2020-09-08)
New Features
- Connection strings used with
from_connection_string
methods now supports using theSharedAccessSignature
key in leiu ofsharedaccesskey
andsharedaccesskeyname
, taking the string of the properly constructed token as value.
5.2.0b1 (2020-07-06)
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.
- Added enum class
azure.eventhub.LoadBalancingStrategy
forload_balancing_strategy
.
5.1.0 (2020-05-04)
New Features
EventHubProducerClient.send_batch
accepts either anEventDataBatch
or a finite list ofEventData
. #9181- Added enqueueTime to span links of distributed tracing. #9599
Bug fixes
- Fixed a bug that turned
azure.eventhub.EventhubConsumerClient
into an exclusive receiver when it has no checkpoint store. #11181 - Updated uAMQP dependency to 1.2.7.
- Fixed bug in setting certificate of tlsio on MacOS. #7201
- Fixed bug that caused segmentation fault in network tracing on MacOS when setting
logging_enable
toTrue
inEventHubConsumerClient
andEventHubProducerClient
.
5.1.0b1 (2020-04-06)
New Features
- Added
EventHubConsumerClient.receive_batch()
to receive and process events in batches instead of one by one. #9184 EventHubConsumerCliuent.receive()
has a new parammax_wait_time
.on_event
is called everymax_wait_time
when no events are received andmax_wait_time
is notNone
or 0.- Param event of
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. #10408
5.0.1 (2020-03-09)
Bug fixes
- Fixed a bug that swallowed errors when receiving events with
azure.eventhub.EventHubConsumerClient
#9660 - Fixed a bug that caused
get_eventhub_properties
,get_partition_ids
, andget_partition_properties
to raise an error on Azure Stack #9920
5.0.0 (2020-01-13)
Breaking changes
EventData
- Removed deprecated property
application_properties
and deprecated methodencode_message()
.
- Removed deprecated property
EventHubConsumerClient
on_error
would be called whenEventHubConsumerClient
failed to claim ownership of partitions.on_partition_close
andon_partition_initialize
would be called in the case of exceptions raised byon_event
callback.EventHubConsumerClient
would close and re-open the internal partition receiver in this case.
- Default starting position from where
EventHubConsumerClient
should resume receiving after recovering from an error has been re-prioritized.- If there is checkpoint, it will resume from the checkpoint.
- If there is no checkpoint but
starting_position
is provided, it will resume fromstarting_posititon
. - If there is no checkpoint or
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.- The in-memory checkpoints would be used for
EventHubConsumerClient
receiving recovering.
- The in-memory checkpoints would be used for
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.
5.0.0b6 (2019-12-03)
Breaking changes
- All exceptions should now be imported from
azure.eventhub.exceptions
. - Introduced separate
EventHubSharedKeyCredential
objects for synchronous and asynchronous operations. For async, import the credentials object from theazure.eventhub.aio
namespace. EventData
- Renamed property
application_properties
toproperties
. EventData
no longer has attributelast_enqueued_event_properties
- use this onPartitionContext
instead.
- Renamed property
EvenDataBatch
EventDataBatch.try_add
has been renamed toEventDataBatch.add
.- Renamed property
size
tosize_in_bytes
. - Renamed attribute
max_size
tomax_size_in_bytes
.
EventHubConsumerClient
andEventHubProducerClient
- Renamed method
get_properties
toget_eventhub_properties
. - Renamed parameters in constructor:
host
tofully_qualified_namespace
,event_hub_path
toeventhub_name
. - Renamed parameters in
get_partition_properties
:partition
topartition_id
. - Renamed parameter
consumer_group_name
toconsumer_group
and moved that parameter fromreceive
method to the constructor ofEventHubConsumerClient
. - Renamed parameter
initial_event_position
tostarting_position
on thereceive
method ofEventHubConsumerClient
. - Renamed parameter
event_hub_path
toeventhub_name
in constructor andfrom_connection_string
method of the client object. EventHubProducerClient.send
has been renamed tosend_batch
which will only acceptEventDataBatch
object as input.EventHubProducerClient.create_batch
now also takes thepartition_id
andpartition_key
as optional parameters (which are no longer specified at send).
- Renamed method
- Renamed module
PartitionManager
toCheckpointStore
. - Receive event callback parameter has been renamed to
on_event
and now operates on a single event rather than a list of events. - Removed class
EventPostition
.- The
starting_position
parameter of thereceive
method accepts offset(str
), sequence number(int
), datetime (datetime.datetime
) ordict
of these types. - The
starting_position_inclusive
parameter of thereceive
method acceptsbool
ordict
indicating whether the given event position is inclusive or not.
- The
PartitionContext
no longer has attributeowner_id
.PartitionContext
now has attributelast_enqueued_event_properties
which is populated iftrack_last_enqueued_event_properties
is set toTrue
in thereceive
method.
New features
- Added new parameter
idle_timeout
in construct andfrom_connection_string
toEventHubConsumerClient
andEventHubProducerClient
after which the underlying connection will close if there is no further activity.
5.0.0b5 (2019-11-04)
Breaking changes
EventHubClient
,EventHubConsumer
andEventHubProducer
has been removed. UseEventHubProducerClient
andEventHubConsumerClient
instead.- Construction of both objects is the same as it was for the previous client.
- Introduced
EventHubProducerClient
as substitution forEventHubProducer
.EventHubProducerClient
supports sending events to different partitions.
- Introduced
EventHubConsumerClient
as substitution forEventHubConsumer
.EventHubConsumerClient
supports receiving events from single/all partitions.- There are no longer methods which directly return
EventData
, all receiving is done via callback method:on_events
.
EventHubConsumerClient
has taken on the responsibility ofEventProcessor
.EventHubConsumerClient
now acceptsPartitionManager
to do load-balancing and checkpoint.
- Replaced
PartitionProcessor
by four independent callback methods accepted by thereceive
method onEventHubConsumerClient
.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.
- Some modules and classes that were importable from several different places have been removed:
azure.eventhub.common
has been removed. Import fromazure.eventhub
instead.azure.eventhub.client_abstract
has been removed. Useazure.eventhub.EventHubProducerClient
orazure.eventhub.EventHubConsumerClient
instead.azure.eventhub.client
has been removed. Useazure.eventhub.EventHubProducerClient
orazure.eventhub.EventHubConsumerClient
instead.azure.eventhub.producer
has been removed. Useazure.eventhub.EventHubProducerClient
instead.azure.eventhub.consumer
has been removed. Useazure.eventhub.EventHubConsumerClient
instead.azure.eventhub.aio.client_async
has been removed. Useazure.eventhub.aio.EventHubProducerClient
orazure.eventhub.aio.EventHubConsumerClient
instead.azure.eventhub.aio.producer_async
has been removed. Useazure.eventhub.aio.EventHubProducerClient
instead.azure.eventhub.aio.consumer_async
has been removed. Useazure.eventhub.aio.EventHubConsumerClient
instead.azure.eventhub.aio.event_processor.event_processor
has been removed. Useazure.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 fromazure.eventhub.aio
instead.azure.eventhub.aio.event_processor.partition_context
has been removed. Import fromazure.eventhub.aio
instead.azure.eventhub.aio.event_processor.sample_partition_manager
has been removed.
Bug fixes
- Fixed bug in user-agent string not being parsed.
5.0.0b4 (2019-10-08)
New features
- Added support for tracing (issue #7153).
- Added the capability of tracking last enqueued event properties of the partition to
EventHubConsumer
.- Added new boolean type parameter
track_last_enqueued_event_properties
in methodEventHubClient.create_consumer()
. - Added new property
last_enqueued_event_properties
ofEventHubConsumer
which contains sequence_number, offset, enqueued_time and retrieval_time information. - By default the capability is disabled as it will cost extra bandwidth for transferring more information if turned on.
- Added new boolean type parameter
Breaking changes
- Removed support for IoT Hub direct connection.
- EventHubs compatible connection string of an IotHub can be used to create
EventHubClient
and read properties or events from an IoT Hub.
- EventHubs compatible connection string of an IotHub can be used to create
- Removed support for sending EventData to IoT Hub.
- Removed parameter
exception
in methodclose()
ofEventHubConsumer
andEventHubProcuer
. - Updated uAMQP dependency to 1.2.3.
5.0.0b3 (2019-09-10)
New features
- Added support for automatic load balancing among multiple
EventProcessor
. - Added
BlobPartitionManager
which implementsPartitionManager
.- Azure Blob Storage is applied for storing data used by
EventProcessor
. - Packaged separately as a plug-in to
EventProcessor
. - For details, please refer to Azure Blob Storage Partition Manager.
- Azure Blob Storage is applied for storing data used by
- Added property
system_properties
onEventData
.
Breaking changes
- Removed constructor method of
PartitionProcessor
. For initialization please implement the methodinitialize
. - Replaced
CheckpointManager
byPartitionContext
.PartitionContext
has partition context information and methodupdate_checkpoint
.
- Updated all methods of
PartitionProcessor
to includePartitionContext
as part of the arguments. - Updated accessibility of class members in
EventHub/EventHubConsumer/EventHubProducer
to be private. - Moved
azure.eventhub.eventprocessor
underaio
package, which now becomesazure.eventhub.aio.eventprocessor
.
5.0.0b2 (2019-08-06)
New features
- Added method
create_batch
on theEventHubProducer
to create anEventDataBatch
that can then be used to add events until the maximum size is reached.- This batch object can then be used in the
send()
method to send all the added events to Event Hubs. - This allows publishers to build batches without the possibility of encountering the error around the message size exceeding the supported limit when sending events.
- It also allows publishers with bandwidth concerns to control the size of each batch published.
- This batch object can then be used in the
- Added new configuration parameters for exponential delay between retry operations.
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.
- Added support for context manager on
EventHubClient
. - Added new error type
OperationTimeoutError
for send operation. - Introduced a new class
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 ofEventProcessor
. 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 byEventProcessor
, please read the new features for details.- Replaced
max_retries
configuration parameter of the EventHubClient withretry_total
.
5.0.0b1 (2019-06-25)
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
- Added new configuration parameters for creating EventHubClient.
credential
: The credential object used for authentication which implementsTokenCredential
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.- for detailed information about the configuration parameters, please read the reference documentation.
- Added new methods
get_partition_properties
andget_partition_ids
to EventHubClient. - Added support for http proxy.
- Added support for authentication using azure-identity credential.
- Added support for transport using AMQP over WebSocket.
Breaking changes
- New error hierarchy
azure.error.EventHubError
azure.error.ConnectionLostError
azure.error.ConnectError
azure.error.AuthenticationError
azure.error.EventDataError
azure.error.EventDataSendError
- Renamed Sender/Receiver to EventHubProducer/EventHubConsumer.
- Renamed
add_sender
tocreate_producer
andadd_receiver
tocreate_consumer
in EventHubClient. - EventHubConsumer is now iterable.
- Renamed
- Rename class azure.eventhub.Offset to azure.eventhub.EventPosition.
- Rename method
get_eventhub_info
toget_properties
of EventHubClient. - Reorganized connection management, EventHubClient is no longer responsible for opening/closing EventHubProducer/EventHubConsumer.
- Each EventHubProducer/EventHubConsumer is responsible for its own connection management.
- Added support for context manager on EventHubProducer and EventHubConsumer.
- Reorganized async APIs into "azure.eventhub.aio" namespace and rename to drop the "_async" suffix.
- Updated uAMQP dependency to 1.2.
1.3.1 (2019-02-28)
BugFixes
- Fixed bug where datetime offset filter was using a local timestamp rather than UTC.
- Fixed stackoverflow error in continuous connection reconnect attempts.
1.3.0 (2019-01-29)
BugFixes
- Added support for auto reconnect on token expiration and other auth errors (issue #89).
Features
- Added ability to create ServiceBusClient from an existing SAS auth token, including providing a function to auto-renew that token on expiry.
- Added support for storing a custom EPH context value in checkpoint (PR #84, thanks @konstantinmiller)
1.2.0 (2018-11-29)
- Support for Python 2.7 in azure.eventhub module (azure.eventprocessorhost will not support Python 2.7).
- Parse EventData.enqueued_time as a UTC timestamp (issue #72, thanks @vjrantal)
1.1.1 (2018-10-03)
- Fixed bug in Azure namespace package.
1.1.0 (2018-09-21)
-
Changes to
AzureStorageCheckpointLeaseManager
parameters to support other connection options (issue #61):- The
storage_account_name
,storage_account_key
andlease_container_name
arguments are now optional keyword arguments. - Added a
sas_token
argument that must be specified withstorage_account_name
in place ofstorage_account_key
. - Added an
endpoint_suffix
argument to support storage endpoints in National Clouds. - Added a
connection_string
argument that, if specified, overrides all other endpoint arguments. - The
lease_container_name
argument now defaults to"eph-leases"
if not specified.
- The
-
Fix for clients failing to start if run called multipled times (issue #64).
-
Added convenience methods
body_as_str
andbody_as_json
to EventData object for easier processing of message data.
1.0.0 (2018-08-22)
- API stable.
- Renamed internal
_async
module toasync_ops
for docs generation. - Added optional
auth_timeout
parameter toEventHubClient
andEventHubClientAsync
to configure how long to allow for token negotiation to complete. Default is 60 seconds. - Added optional
send_timeout
parameter toEventHubClient.add_sender
andEventHubClientAsync.add_async_sender
to determine the timeout for Events to be successfully sent. Default value is 60 seconds. - Reformatted logging for performance.
0.2.0 (2018-08-06)
-
Stability improvements for EPH.
-
Updated uAMQP version.
-
Added new configuration options for Sender and Receiver;
keep_alive
andauto_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
0.2.0rc2 (2018-07-29)
- Breaking change
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
. - Each sender/receiver will now run in its own independent connection.
- Updated uAMQP dependency to 0.2.0
- Fixed issue with IoTHub clients not being able to retrieve partition information.
- Added support for HTTP proxy settings to both EventHubClient and EPH.
- Added error handling policy to automatically reconnect on retryable error.
- Added keep-alive thread for maintaining an unused connection.
0.2.0rc1 (2018-07-06)
- Breaking change Restructured library to support Python 3.7. Submodule
async
has been renamed and all classes from this module can now be imported from azure.eventhub directly. - Breaking change Removed optional
callback
argument fromReceiver.receive
andAsyncReceiver.receive
. - Breaking change
EventData.properties
has been renamed toEventData.application_properties
. This removes the potential for messages to be processed via callback for not yet returned in the batch. - Updated uAMQP dependency to v0.1.0
- Added support for constructing IoTHub connections.
- Fixed memory leak in receive operations.
- Dropped Python 2.7 wheel support.
0.2.0b2 (2018-05-29)
- Added
namespace_suffix
to EventHubConfig() to support national clouds. - Added
device_id
attribute to EventData to support IoT Hub use cases. - Added message header to workaround service bug for PartitionKey support.
- Updated uAMQP dependency to vRC1.
0.2.0b1 (2018-04-20)
- Updated uAMQP to latest version.
- Further testing and minor bug fixes.
0.2.0a2 (2018-04-02)
- Updated uAQMP dependency.
0.2.0a1 (unreleased)
- Swapped out Proton dependency for uAMQP.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for azure_eventhub-5.8.0a1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 271c0ddd209e455d96366637af651160b7c067aebfd546519ca13980bb257735 |
|
MD5 | 838270b06db9dfaafce0bc3ecec5fa7a |
|
BLAKE2b-256 | 3ed2e559fe552ed0db72f8905bd68958602c524891fdf92bfaec30ff1077aabb |