EventHubProducerClient Class
The EventHubProducerClient class defines a high level interface for sending events to the Azure Event Hubs service.
- Inheritance
-
azure.eventhub._client_base.ClientBaseEventHubProducerClient
Constructor
EventHubProducerClient(fully_qualified_namespace: str, eventhub_name: str, credential: CredentialTypes, *, buffered_mode: Literal[False] = False, **kwargs: Any)
Parameters
Name | Description |
---|---|
fully_qualified_namespace
Required
|
The fully qualified host name for the Event Hubs namespace. This is likely to be similar to .servicebus.windows.net |
eventhub_name
Required
|
The path of the specific Event Hub to connect the client to. |
credential
Required
|
The credential object used for authentication which implements a particular interface for getting tokens. It accepts EventHubSharedKeyCredential, or credential objects generated by the azure-identity library and objects that implement the *get_token(self, scopes) method. |
Keyword-Only Parameters
Name | Description |
---|---|
buffered_mode
|
If True, the producer client will collect events in a buffer, efficiently batch, then publish. Default is False. |
buffer_concurrency
|
The ThreadPoolExecutor to be used for publishing events or the number of workers for the ThreadPoolExecutor. Default is None and a ThreadPoolExecutor with the default number of workers will be created per https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor |
on_success
|
The callback to be called once a batch has been successfully published. The callback takes two parameters:
The callback function should be defined like: on_success(events, partition_id). It is required when buffered_mode is True while optional if buffered_mode is False. |
on_error
|
The callback to be called once a batch has failed to be published. The callback function should be defined like: on_error(events, partition_id, error), where:
If buffered_mode is False, on_error callback is optional and errors will be handled as follows:
If buffered_mode is True, on_error callback is required and errors will be handled as follows:
|
max_buffer_length
|
Buffered mode only. The total number of events per partition that can be buffered before a flush will be triggered. The default value is 1500 in buffered mode. |
max_wait_time
|
Buffered mode only. The amount of time to wait for a batch to be built with events in the buffer before publishing. The default value is 1 in buffered mode. |
logging_enable
|
Whether to output network trace logs to the logger. Default is False. |
auth_timeout
|
The time in seconds to wait for a token to be authorized by the service. The default value is 60 seconds. If set to 0, no timeout will be enforced from the client. |
user_agent
|
If specified, this will be added in front of the user agent string. |
retry_total
|
The total number of attempts to redo a failed operation when an error occurs. Default value is 3. |
retry_backoff_factor
|
A backoff factor to apply between attempts after the second try (most errors are resolved immediately by a second try without a delay). In fixed mode, retry policy will always sleep for {backoff factor}. In 'exponential' mode, retry policy will sleep for: {backoff factor} * (2 ** ({number of total retries} - 1)) seconds. If the backoff_factor is 0.1, then the retry will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. The default value is 0.8. |
retry_backoff_max
|
The maximum back off time. Default value is 120 seconds (2 minutes). |
retry_mode
|
The delay behavior between retry attempts. Supported values are 'fixed' or 'exponential', where default is 'exponential'. |
idle_timeout
|
Timeout, in seconds, after which this client will close the underlying connection if there is no activity. By default the value is None, meaning that the client will not shutdown due to inactivity unless initiated by the service. |
transport_type
|
The type of transport protocol that will be used for communicating with the Event Hubs service. Default is TransportType.Amqp in which case port 5671 is used. If the port 5671 is unavailable/blocked in the network environment, TransportType.AmqpOverWebsocket could be used instead which uses port 443 for communication. |
http_proxy
|
HTTP proxy settings. This must be a dictionary with the following keys: 'proxy_hostname' (str value) and 'proxy_port' (int value). Additionally the following keys may also be present: 'username', 'password'. |
custom_endpoint_address
|
The custom endpoint address to use for establishing a connection to the Event Hubs service, allowing network requests to be routed through any application gateways or other paths needed for the host environment. Default is None. The format would be like "sb://<custom_endpoint_hostname>:<custom_endpoint_port>". If port is not specified in the custom_endpoint_address, by default port 443 will be used. |
connection_verify
|
Path to the custom CA_BUNDLE file of the SSL certificate which is used to authenticate the identity of the connection endpoint. Default is None in which case certifi.where() will be used. |
uamqp_transport
|
Whether to use the uamqp library as the underlying transport. The default value is False and the Pure Python AMQP library will be used as the underlying transport. |
socket_timeout
|
The time in seconds that the underlying socket on the connection should wait when sending and receiving data before timing out. The default value is 0.2 for TransportType.Amqp and 1 for TransportType.AmqpOverWebsocket. If EventHubsConnectionError errors are occurring due to write timing out, a larger than default value may need to be passed in. This is for advanced usage scenarios and ordinarily the default value should be sufficient. |
Examples
Create a new instance of the EventHubProducerClient.
import os
from azure.eventhub import EventHubProducerClient, EventHubSharedKeyCredential
fully_qualified_namespace = os.environ['EVENT_HUB_HOSTNAME']
eventhub_name = os.environ['EVENT_HUB_NAME']
shared_access_policy = os.environ['EVENT_HUB_SAS_POLICY']
shared_access_key = os.environ['EVENT_HUB_SAS_KEY']
credential = EventHubSharedKeyCredential(shared_access_policy, shared_access_key)
producer = EventHubProducerClient(
fully_qualified_namespace=fully_qualified_namespace,
eventhub_name=eventhub_name, # EventHub name should be specified if it doesn't show up in connection string.
credential=credential
)
Methods
close |
Close the Producer client underlying AMQP connection and links. |
create_batch |
Create an EventDataBatch object with the max size of all content being constrained by max_size_in_bytes. The max_size_in_bytes should be no greater than the max allowed message size defined by the service. |
flush |
Buffered mode only. Flush events in the buffer to be sent immediately if the client is working in buffered mode. |
from_connection_string |
Create an EventHubProducerClient from a connection string. |
get_buffered_event_count |
The number of events that are buffered and waiting to be published for a given partition. Returns None in non-buffered mode. NOTE: The event buffer is processed in a background thread, therefore the number of events in the buffer reported by this API should be considered only an approximation and is only recommend for use in debugging. For a partition ID that has no events buffered, 0 will be returned regardless of whether that partition ID actually exists within the Event Hub. |
get_eventhub_properties |
Get properties of the Event Hub. Keys in the returned dictionary include:
|
get_partition_ids |
Get partition IDs of the Event Hub. |
get_partition_properties |
Get properties of the specified partition. Keys in the properties dictionary include:
|
send_batch |
Sends a batch of event data. By default, the method will block until acknowledgement is received or operation times out. If the EventHubProducerClient is configured to run in buffered mode, the method will try enqueuing the events into buffer within the given time if specified and return. The producer will do automatic sending in the background in buffered mode. If buffered_mode is False, on_error callback is optional and errors will be handled as follows:
If buffered_mode is True, on_error callback is required and errors will be handled as follows:
In buffered mode, sending a batch will remain intact and sent as a single unit. The batch will not be rearranged. This may result in inefficiency of sending events. If you're sending a finite list of EventData or AmqpAnnotatedMessage and you know it's within the event hub frame size limit, you can send them with a send_batch call. Otherwise, use create_batch to create EventDataBatch and add either EventData or AmqpAnnotatedMessage into the batch one by one until the size limit, and then call this method to send out the batch. |
send_event |
Sends an event data. By default, the method will block until acknowledgement is received or operation times out. If the EventHubProducerClient is configured to run in buffered mode, the method will try enqueuing the events into buffer within the given time if specified and return. The producer will do automatic sending in the background in buffered mode. If buffered_mode is False, on_error callback is optional and errors will be handled as follows: * If an on_error callback is passed during the producer client instantiation,
If buffered_mode is True, on_error callback is required and errors will be handled as follows: * If events fail to enqueue within the given timeout, then an error will be directly raised.
|
close
Close the Producer client underlying AMQP connection and links.
close(*, flush: bool = True, timeout: float | None = None, **kwargs: Any) -> None
Keyword-Only Parameters
Name | Description |
---|---|
flush
|
Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True. Default value: True
|
timeout
|
Buffered mode only. Timeout to close the producer. Default is None which means no timeout. |
Returns
Type | Description |
---|---|
Exceptions
Type | Description |
---|---|
If an error occurred when flushing the buffer if flush is set to True or closing the underlying AMQP connections in buffered mode. |
Examples
Close down the client.
import os
from azure.eventhub import EventHubProducerClient, EventData
from azure.identity import DefaultAzureCredential
eventhub_fully_qualified_namespace = os.environ["EVENT_HUB_HOSTNAME"]
eventhub_name = os.environ['EVENT_HUB_NAME']
producer = EventHubProducerClient(
fully_qualified_namespace=eventhub_fully_qualified_namespace,
eventhub_name=eventhub_name, # EventHub name should be specified if it doesn't show up in connection string.
credential=DefaultAzureCredential()
)
try:
event_data_batch = producer.create_batch()
while True:
try:
event_data_batch.add(EventData('Message inside EventBatchData'))
except ValueError:
# EventDataBatch object reaches max_size.
# New EventDataBatch object can be created here to send more data
break
producer.send_batch(event_data_batch)
finally:
# Close down the producer handler.
producer.close()
create_batch
Create an EventDataBatch object with the max size of all content being constrained by max_size_in_bytes.
The max_size_in_bytes should be no greater than the max allowed message size defined by the service.
create_batch(*, partition_id: str | None = None, partition_key: str | None = None, max_size_in_bytes: int | None = None, **kwargs: Any) -> EventDataBatch
Keyword-Only Parameters
Name | Description |
---|---|
partition_id
Required
|
|
partition_key
Required
|
|
max_size_in_bytes
Required
|
|
Returns
Type | Description |
---|---|
Exceptions
Type | Description |
---|---|
If an error occurred when flushing the buffer if flush is set to True or closing the underlying AMQP connections in buffered mode. |
Examples
Create EventDataBatch object within limited size
event_data_batch = producer.create_batch()
while True:
try:
event_data_batch.add(EventData('Message inside EventBatchData'))
except ValueError:
# The EventDataBatch object reaches its max_size.
# You can send the full EventDataBatch object and create a new one here.
break
flush
Buffered mode only. Flush events in the buffer to be sent immediately if the client is working in buffered mode.
flush(*, timeout: float | None = None, **kwargs: Any) -> None
Keyword-Only Parameters
Name | Description |
---|---|
timeout
|
Timeout to flush the buffered events, default is None which means no timeout. |
Returns
Type | Description |
---|---|
Exceptions
Type | Description |
---|---|
If the producer fails to flush the buffer within the given timeout in buffered mode. |
from_connection_string
Create an EventHubProducerClient from a connection string.
from_connection_string(conn_str: str, *, eventhub_name: str | None = None, buffered_mode: Literal[False] = False, **kwargs: Any) -> EventHubProducerClient
Parameters
Name | Description |
---|---|
conn_str
Required
|
The connection string of an Event Hub. |
Keyword-Only Parameters
Name | Description |
---|---|
eventhub_name
|
The path of the specific Event Hub to connect the client to. |
buffered_mode
|
If True, the producer client will collect events in a buffer, efficiently batch, then publish. Default is False. |
buffer_concurrency
|
The ThreadPoolExecutor to be used for publishing events or the number of workers for the ThreadPoolExecutor. Default is None and a ThreadPoolExecutor with the default number of workers will be created per https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor |
on_success
|
The callback to be called once a batch has been successfully published. The callback takes two parameters:
The callback function should be defined like: on_success(events, partition_id). Required when buffered_mode is True while optional if buffered_mode is False. |
on_error
|
The callback to be called once a batch has failed to be published. Required when in buffered_mode is True while optional if buffered_mode is False. The callback function should be defined like: on_error(events, partition_id, error), where:
If buffered_mode is False, on_error callback is optional and errors will be handled as follows:
If buffered_mode is True, on_error callback is required and errors will be handled as follows:
|
max_buffer_length
|
Buffered mode only. The total number of events per partition that can be buffered before a flush will be triggered. The default value is 1500 in buffered mode. |
max_wait_time
|
Buffered mode only. The amount of time to wait for a batch to be built with events in the buffer before publishing. The default value is 1 in buffered mode. |
logging_enable
|
Whether to output network trace logs to the logger. Default is False. |
http_proxy
|
HTTP proxy settings. This must be a dictionary with the following keys: 'proxy_hostname' (str value) and 'proxy_port' (int value). Additionally the following keys may also be present: 'username', 'password'. |
auth_timeout
|
The time in seconds to wait for a token to be authorized by the service. The default value is 60 seconds. If set to 0, no timeout will be enforced from the client. Default value: 60
|
user_agent
|
If specified, this will be added in front of the user agent string. |
retry_total
|
The total number of attempts to redo a failed operation when an error occurs. Default value is 3. Default value: 3
|
retry_backoff_factor
|
A backoff factor to apply between attempts after the second try (most errors are resolved immediately by a second try without a delay). In fixed mode, retry policy will always sleep for {backoff factor}. In 'exponential' mode, retry policy will sleep for: {backoff factor} * (2 ** ({number of total retries} - 1)) seconds. If the backoff_factor is 0.1, then the retry will sleep for [0.0s, 0.2s, 0.4s, ...] between retries. The default value is 0.8. Default value: 0.8
|
retry_backoff_max
|
The maximum back off time. Default value is 120 seconds (2 minutes). Default value: 120
|
retry_mode
|
The delay behavior between retry attempts. Supported values are 'fixed' or 'exponential', where default is 'exponential'. Default value: exponential
|
idle_timeout
|
Timeout, in seconds, after which this client will close the underlying connection if there is no activity. By default the value is None, meaning that the client will not shutdown due to inactivity unless initiated by the service. |
transport_type
|
The type of transport protocol that will be used for communicating with the Event Hubs service. Default is TransportType.Amqp in which case port 5671 is used. If the port 5671 is unavailable/blocked in the network environment, TransportType.AmqpOverWebsocket could be used instead which uses port 443 for communication. Default value: TransportType.Amqp
|
http_proxy
|
HTTP proxy settings. This must be a dictionary with the following keys: 'proxy_hostname' (str value) and 'proxy_port' (int value). Additionally the following keys may also be present: 'username', 'password'. |
custom_endpoint_address
|
The custom endpoint address to use for establishing a connection to the Event Hubs service, allowing network requests to be routed through any application gateways or other paths needed for the host environment. Default is None. The format would be like "sb://<custom_endpoint_hostname>:<custom_endpoint_port>". If port is not specified in the custom_endpoint_address, by default port 443 will be used. |
connection_verify
|
Path to the custom CA_BUNDLE file of the SSL certificate which is used to authenticate the identity of the connection endpoint. Default is None in which case certifi.where() will be used. |
uamqp_transport
|
Whether to use the uamqp library as the underlying transport. The default value is False and the Pure Python AMQP library will be used as the underlying transport. |
Returns
Type | Description |
---|---|
An instance of the EventHubProducerClient. |
Exceptions
Type | Description |
---|---|
If an error occurred when flushing the buffer if flush is set to True or closing the underlying AMQP connections in buffered mode. |
Examples
Create a new instance of the EventHubProducerClient from connection string.
import os
from azure.identity import DefaultAzureCredential
eventhub_fully_qualified_namespace = os.environ["EVENT_HUB_HOSTNAME"]
eventhub_name = os.environ['EVENT_HUB_NAME']
producer = EventHubProducerClient(
fully_qualified_namespace=eventhub_fully_qualified_namespace,
eventhub_name=eventhub_name, # EventHub name should be specified if it doesn't show up in connection string.
credential=DefaultAzureCredential()
)
get_buffered_event_count
The number of events that are buffered and waiting to be published for a given partition. Returns None in non-buffered mode. NOTE: The event buffer is processed in a background thread, therefore the number of events in the buffer reported by this API should be considered only an approximation and is only recommend for use in debugging. For a partition ID that has no events buffered, 0 will be returned regardless of whether that partition ID actually exists within the Event Hub.
get_buffered_event_count(partition_id: str) -> int | None
Parameters
Name | Description |
---|---|
partition_id
Required
|
The target partition ID. |
Keyword-Only Parameters
Name | Description |
---|---|
flush
|
Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True. Default value: True
|
timeout
|
Buffered mode only. Timeout to close the producer. Default is None which means no timeout. |
Returns
Type | Description |
---|---|
int,
|
Exceptions
Type | Description |
---|---|
If an error occurred when flushing the buffer if flush is set to True or closing the underlying AMQP connections in buffered mode. |
get_eventhub_properties
Get properties of the Event Hub.
Keys in the returned dictionary include:
eventhub_name (str)
created_at (UTC datetime.datetime)
partition_ids (list[str])
get_eventhub_properties() -> Dict[str, Any]
Keyword-Only Parameters
Name | Description |
---|---|
flush
|
Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True. Default value: True
|
timeout
|
Buffered mode only. Timeout to close the producer. Default is None which means no timeout. |
Returns
Type | Description |
---|---|
A dictionary containing eventhub properties. |
Exceptions
Type | Description |
---|---|
get_partition_ids
Get partition IDs of the Event Hub.
get_partition_ids() -> List[str]
Keyword-Only Parameters
Name | Description |
---|---|
flush
|
Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True. Default value: True
|
timeout
|
Buffered mode only. Timeout to close the producer. Default is None which means no timeout. |
Returns
Type | Description |
---|---|
A list of partition IDs. |
Exceptions
Type | Description |
---|---|
get_partition_properties
Get properties of the specified partition.
Keys in the properties dictionary include:
eventhub_name (str)
id (str)
beginning_sequence_number (int)
last_enqueued_sequence_number (int)
last_enqueued_offset (str)
last_enqueued_time_utc (UTC datetime.datetime)
is_empty (bool)
get_partition_properties(partition_id: str) -> Dict[str, Any]
Parameters
Name | Description |
---|---|
partition_id
Required
|
The target partition ID. |
Keyword-Only Parameters
Name | Description |
---|---|
flush
|
Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True. Default value: True
|
timeout
|
Buffered mode only. Timeout to close the producer. Default is None which means no timeout. |
Returns
Type | Description |
---|---|
A dictionary of partition properties. |
Exceptions
Type | Description |
---|---|
send_batch
Sends a batch of event data. By default, the method will block until acknowledgement is received or operation times out. If the EventHubProducerClient is configured to run in buffered mode, the method will try enqueuing the events into buffer within the given time if specified and return. The producer will do automatic sending in the background in buffered mode.
If buffered_mode is False, on_error callback is optional and errors will be handled as follows:
If an on_error callback is passed during the producer client instantiation,
then error information will be passed to the on_error callback, which will then be called.
If an on_error callback is not passed in during client instantiation,
then the error will be raised by default.
If buffered_mode is True, on_error callback is required and errors will be handled as follows:
If events fail to enqueue within the given timeout, then an error will be directly raised.
If events fail to send after enqueuing successfully, the on_error callback will be called.
In buffered mode, sending a batch will remain intact and sent as a single unit. The batch will not be rearranged. This may result in inefficiency of sending events.
If you're sending a finite list of EventData or AmqpAnnotatedMessage and you know it's within the event hub frame size limit, you can send them with a send_batch call. Otherwise, use create_batch to create EventDataBatch and add either EventData or AmqpAnnotatedMessage into the batch one by one until the size limit, and then call this method to send out the batch.
send_batch(event_data_batch: EventDataBatch | List[EventData | AmqpAnnotatedMessage], *, timeout: float | None = None, partition_key: str | None = None, partition_id: str | None = None, **kwargs: Any) -> None
Parameters
Name | Description |
---|---|
event_data_batch
Required
|
The EventDataBatch object to be sent or a list of EventData to be sent in a batch. All EventData or AmqpAnnotatedMessage in the list or EventDataBatch will land on the same partition. |
Keyword-Only Parameters
Name | Description |
---|---|
timeout
|
The maximum wait time to send the event data in non-buffered mode or the maximum wait time to enqueue the event data into the buffer in buffered mode. In non-buffered mode, the default wait time specified when the producer was created will be used. In buffered mode, the default wait time is None. |
partition_id
|
The specific partition ID to send to. Default is None, in which case the service will assign to all partitions using round-robin. A TypeError will be raised if partition_id is specified and event_data_batch is an EventDataBatch because EventDataBatch itself has partition_id. |
partition_key
|
With the given partition_key, event data will be sent to a particular partition of the Event Hub decided by the service. A TypeError will be raised if partition_key is specified and event_data_batch is an EventDataBatch because EventDataBatch itself has partition_key. If both partition_id and partition_key are provided, the partition_id will take precedence. WARNING: Setting partition_key of non-string value on the events to be sent is discouraged as the partition_key will be ignored by the Event Hub service and events will be assigned to all partitions using round-robin. Furthermore, there are SDKs for consuming events which expect partition_key to only be string type, they might fail to parse the non-string value. |
Returns
Type | Description |
---|---|
Exceptions
Type | Description |
---|---|
If the value specified by the timeout parameter elapses before the event can be sent in non-buffered mode or the events can not be enqueued into the buffered in buffered mode. |
Examples
Sends event data
with producer:
event_data_batch = producer.create_batch()
while True:
try:
event_data_batch.add(EventData('Message inside EventBatchData'))
except ValueError:
# EventDataBatch object reaches max_size.
# New EventDataBatch object can be created here to send more data
break
producer.send_batch(event_data_batch)
send_event
Sends an event data. By default, the method will block until acknowledgement is received or operation times out. If the EventHubProducerClient is configured to run in buffered mode, the method will try enqueuing the events into buffer within the given time if specified and return. The producer will do automatic sending in the background in buffered mode.
If buffered_mode is False, on_error callback is optional and errors will be handled as follows: * If an on_error callback is passed during the producer client instantiation,
then error information will be passed to the *on_error* callback, which will then be called.
* If an *on_error* callback is not passed in during client instantiation,
then the error will be raised by default.
If buffered_mode is True, on_error callback is required and errors will be handled as follows: * If events fail to enqueue within the given timeout, then an error will be directly raised.
* If events fail to send after enqueuing successfully, the *on_error* callback will be called.
send_event(event_data: EventData | AmqpAnnotatedMessage, *, timeout: float | None = None, partition_id: str | None = None, partition_key: str | None = None, **kwargs: Any) -> None
Parameters
Name | Description |
---|---|
event_data
Required
|
The EventData object to be sent. |
Keyword-Only Parameters
Name | Description |
---|---|
timeout
|
The maximum wait time to send the event data in non-buffered mode or the maximum wait time to enqueue the event data into the buffer in buffered mode. In non-buffered mode, the default wait time specified when the producer was created will be used. In buffered mode, the default wait time is None indicating that the event will be scheduled to send immediately. |
partition_id
|
The specific partition ID to send to. Default is None, in which case the service will assign to all partitions using round-robin. A TypeError will be raised if partition_id is specified and event_data_batch is an EventDataBatch because EventDataBatch itself has partition_id. |
partition_key
|
With the given partition_key, event data will be sent to a particular partition of the Event Hub decided by the service. A TypeError will be raised if partition_key is specified and event_data_batch is an EventDataBatch because EventDataBatch itself has partition_key. If both partition_id and partition_key are provided, the partition_id will take precedence. WARNING: Setting partition_key of non-string value on the events to be sent is discouraged as the partition_key will be ignored by the Event Hub service and events will be assigned to all partitions using round-robin. Furthermore, there are SDKs for consuming events which expect partition_key to only be string type, they might fail to parse the non-string value. |
Returns
Type | Description |
---|---|
Exceptions
Type | Description |
---|---|
If the value specified by the timeout parameter elapses before the event can be sent in non-buffered mode or the events can be enqueued into the buffered in buffered mode. |
Attributes
total_buffered_event_count
The total number of events that are currently buffered and waiting to be published, across all partitions. Returns None in non-buffered mode. NOTE: The event buffer is processed in a background thread, therefore the number of events in the buffer reported by this API should be considered only an approximation and is only recommend for use in debugging.
Returns
Type | Description |
---|---|
int,
|
Azure SDK for Python