Share via

EventHubProducerClient Class

  • Reference

The EventHubProducerClient class defines a high level interface for sending events to the Azure Event Hubs service.

Inheritance
azure.eventhub.aio._client_base_async.ClientBaseAsync
EventHubProducerClient

Constructor

EventHubProducerClient(fully_qualified_namespace: str, eventhub_name: str, credential: CredentialTypes, *, buffered_mode: bool = False, on_error: Callable[[List[EventData | AmqpAnnotatedMessage], str | None, Exception], Awaitable[None]] | None = None, on_success: Callable[[List[EventData | AmqpAnnotatedMessage], str | None], Awaitable[None]] | None = None, max_buffer_length: int | None = None, max_wait_time: float | None = None, **kwargs: Any)

Parameters

Name Description
fully_qualified_namespace
Required
str

The fully qualified host name for the Event Hubs namespace. This is likely to be similar to .servicebus.windows.net
eventhub_name
Required
str

The path of the specific Event Hub to connect the client to.
credential
Required
AsyncTokenCredential or AzureSasCredential or AzureNamedKeyCredential

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
bool

If True, the producer client will collect events in a buffer, efficiently batch, then publish. Default is False.
on_success
Optional[Callable[[<xref:SendEventTypes>, Optional[str]], Awaitable[None]]]

The callback to be called once a batch has been successfully published. The callback takes two parameters:

  • events: The list of events that have been successfully published
  • partition_id: The partition id that the events in the list have been published to.

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
Optional[Callable[[<xref:SendEventTypes>, Optional[str], Exception], Awaitable[None]]]

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:

  • events: The list of events that failed to be published,
  • partition_id: The partition id that the events in the list have been tried to be published to and
  • error: The exception related to the sending failure.

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.
max_buffer_length
int

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
Optional[float]

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
bool

Whether to output network trace logs to the logger. Default is False.
auth_timeout
float

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
str

If specified, this will be added in front of the user agent string.
retry_total
int

The total number of attempts to redo a failed operation when an error occurs. Default value is 3.
retry_backoff_factor
float

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
float

The maximum back off time. Default value is 120 seconds (2 minutes).
retry_mode
str

The delay behavior between retry attempts. Supported values are 'fixed' or 'exponential', where default is 'exponential'.
idle_timeout
float

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
TransportType

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
dict

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
Optional[str]

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
Optional[str]

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
bool

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
float

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.aio 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']

   producer = EventHubProducerClient(fully_qualified_namespace=fully_qualified_namespace,
                                     eventhub_name=eventhub_name,
                                     credential=EventHubSharedKeyCredential(shared_access_policy, shared_access_key))

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 coroutine, 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:

  • eventhub_name (str)

  • created_at (UTC datetime.datetime)

  • partition_ids (list[str])
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:

  • 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)
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 enqueue the events into local buffer and return. The producer will do automatic sending in the background.

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_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 enqueue the event into local buffer and return. The producer will do automatic batching and sending in the background.

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.

close

Close the Producer client underlying AMQP connection and links.

async close(*, flush: bool = True, timeout: float | None = None, **kwargs: Any) -> None

Keyword-Only Parameters

Name Description
flush
bool

Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True.

Default value: True
timeout
float or None

Buffered mode only. Timeout to close the producer. Default is None which means no timeout.

Returns

Type Description
None

Exceptions

Type Description
EventHubError

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 handler.


   import os
   from azure.eventhub.aio import EventHubProducerClient
   from azure.eventhub import EventData
   from azure.identity.aio import DefaultAzureCredential

   fully_qualified_namespace = os.environ["EVENT_HUB_HOSTNAME"]
   eventhub_name = os.environ['EVENT_HUB_NAME']

   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=DefaultAzureCredential()
   )
   try:
       event_data_batch = await 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
       await producer.send_batch(event_data_batch)
   finally:
       # Close down the producer handler.
       await 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.

async create_batch(*, partition_id: str | None = None, partition_key: str | None = None, max_size_in_bytes: int | None = None) -> EventDataBatch

Keyword-Only Parameters

Name Description
partition_id
Required
partition_key
Required
max_size_in_bytes
Required

Returns

Type Description
EventDataBatch

Exceptions

Type Description
EventHubError

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


       from azure.eventhub import EventData
       event_data_batch = await 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.

async flush(*, timeout: float | None = None, **kwargs: Any) -> None

Keyword-Only Parameters

Name Description
timeout
float or None

Timeout to flush the buffered events, default is None which means no timeout.

Returns

Type Description
None

Exceptions

Type Description
EventDataSendError

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: bool = False, on_success: Callable[[List[EventData | AmqpAnnotatedMessage], str | None], Awaitable[None]] | None = None, on_error: Callable[[List[EventData | AmqpAnnotatedMessage], str | None, Exception], Awaitable[None]] | None = None, max_buffer_length: int | None = None, max_wait_time: float | None = None, logging_enable: bool = False, http_proxy: Dict[str, str | int] | None = None, auth_timeout: float = 60, user_agent: str | None = None, retry_total: int = 3, retry_backoff_factor: float | None = 0.8, retry_backoff_max: float | None = 120, retry_mode: Literal['exponential', 'fixed'] = 'exponential', idle_timeout: float | None = None, transport_type: TransportType = TransportType.Amqp, custom_endpoint_address: str | None = None, connection_verify: str | None = None, uamqp_transport: bool = False, **kwargs: Any) -> EventHubProducerClient

Parameters

Name Description
conn_str
Required
str

The connection string of an Event Hub.

Keyword-Only Parameters

Name Description
eventhub_name
str

The path of the specific Event Hub to connect the client to.
buffered_mode
bool

If True, the producer client will collect events in a buffer, efficiently batch, then publish. Default is False.
on_success
Optional[Callable[[<xref:SendEventTypes>, Optional[str]], Awaitable[None]]]

The callback to be called once a batch has been successfully published. The callback takes two parameters:

  • events: The list of events that have been successfully published
  • partition_id: The partition id that the events in the list have been published to.

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
Optional[Callable[[<xref:SendEventTypes>, Optional[str], Exception], Awaitable[None]]]

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:

  • events: The list of events that failed to be published,
  • partition_id: The partition id that the events in the list have been tried to be published to and
  • error: The exception related to the sending failure.

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.
max_buffer_length
int

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
Optional[float]

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
bool

Whether to output network trace logs to the logger. Default is False.
http_proxy
dict

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
float

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
str

If specified, this will be added in front of the user agent string.
retry_total
int

The total number of attempts to redo a failed operation when an error occurs. Default value is 3.

Default value: 3
retry_backoff_factor
float

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
float

The maximum back off time. Default value is 120 seconds (2 minutes).

Default value: 120
retry_mode
str

The delay behavior between retry attempts. Supported values are 'fixed' or 'exponential', where default is 'exponential'.

Default value: exponential
idle_timeout
float

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
TransportType

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
custom_endpoint_address
Optional[str]

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
Optional[str]

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
bool

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
EventHubProducerClient

An EventHubProducerClient instance.

Exceptions

Type Description
EventHubError

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.eventhub.aio import EventHubProducerClient
   from azure.identity.aio import DefaultAzureCredential

   fully_qualified_namespace = os.environ["EVENT_HUB_HOSTNAME"]
   eventhub_name = os.environ['EVENT_HUB_NAME']
   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=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 coroutine, 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
str

The target partition ID.

Keyword-Only Parameters

Name Description
flush
bool

Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True.

Default value: True
timeout
float or None

Buffered mode only. Timeout to close the producer. Default is None which means no timeout.

Returns

Type Description
int,
None

Exceptions

Type Description
EventHubError

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])

async get_eventhub_properties() -> Dict[str, Any]

Keyword-Only Parameters

Name Description
flush
bool

Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True.

Default value: True
timeout
float or None

Buffered mode only. Timeout to close the producer. Default is None which means no timeout.

Returns

Type Description
dict[str, any]

A dictionary containing information about the Event Hub.

Exceptions

Type Description
EventHubError

get_partition_ids

Get partition IDs of the Event Hub.

async get_partition_ids() -> List[str]

Keyword-Only Parameters

Name Description
flush
bool

Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True.

Default value: True
timeout
float or None

Buffered mode only. Timeout to close the producer. Default is None which means no timeout.

Returns

Type Description
list[str]

A list of partition IDs.

Exceptions

Type Description
EventHubError

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)

async get_partition_properties(partition_id: str) -> Dict[str, Any]

Parameters

Name Description
partition_id
Required
str

The target partition ID.

Keyword-Only Parameters

Name Description
flush
bool

Buffered mode only. If set to True, events in the buffer will be sent immediately. Default is True.

Default value: True
timeout
float or None

Buffered mode only. Timeout to close the producer. Default is None which means no timeout.

Returns

Type Description
dict[str, any]

A dict of partition properties.

Exceptions

Type Description
EventHubError

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 enqueue the events into local buffer and return. The producer will do automatic sending in the background.

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.

async send_batch(event_data_batch: EventDataBatch | List[EventData | AmqpAnnotatedMessage], *, timeout: float | None = None, partition_id: str | None = None, partition_key: str | None = None, **kwargs: Any) -> None

Parameters

Name Description
event_data_batch
Required
Union[EventDataBatch, List[Union[EventData, AmqpAnnotatedMessage]]

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
float

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
str

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
str

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
None

Exceptions

Type Description
AuthenticationError
ConnectError
ConnectionLostError
EventDataError
EventDataSendError
EventHubError
ValueError
TypeError
OperationTimeoutError

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.

Examples

Asynchronously sends event data


       async with producer:
           event_data_batch = await 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
           await 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 enqueue the event into local buffer and return. The producer will do automatic batching and sending in the background.

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.

async 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
Union[EventData, AmqpAnnotatedMessage]

The EventData object to be sent.

Keyword-Only Parameters

Name Description
timeout
float

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
str

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
str

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
None

Exceptions

Type Description
AuthenticationError
ConnectError
ConnectionLostError
EventDataError
EventDataSendError
EventHubError
OperationTimeoutError

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.

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 coroutine, 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,
None