diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/client_async.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/client_async.py index 07f3bf53f87e..4ad5a99f538c 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/client_async.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/client_async.py @@ -251,7 +251,7 @@ def create_producer(self, *, partition_id=None, operation=None, send_timeout=Non queued. Default value is 60 seconds. If set to 0, there will be no timeout. :type send_timeout: float :param loop: An event loop. If not specified the default event loop will be used. - :rtype ~azure.eventhub.aio.producer_async.EventHubProducer + :rtype: ~azure.eventhub.aio.producer_async.EventHubProducer Example: .. literalinclude:: ../examples/async_examples/test_examples_eventhub_async.py diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py index a231e12eb63c..c45cf9a6b283 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py @@ -156,6 +156,7 @@ async def create_batch(self, max_size=None, partition_key=None): """ Create an EventDataBatch object with max size being max_size. The max_size should be no greater than the max allowed message size defined by the service side. + :param max_size: The maximum size of bytes data that an EventDataBatch object can hold. :type max_size: int :param partition_key: With the given partition_key, event data will land to @@ -197,7 +198,8 @@ async def send(self, event_data, *, partition_key=None, timeout=None): :type partition_key: str :param timeout: The maximum wait time to send the event data. If not specified, the default wait time specified when the producer was created will be used. - :type timeout:float + :type timeout: float + :raises: ~azure.eventhub.AuthenticationError, ~azure.eventhub.ConnectError, ~azure.eventhub.ConnectionLostError, ~azure.eventhub.EventDataError, ~azure.eventhub.EventDataSendError, ~azure.eventhub.EventHubError :return: None diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/consumer.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/consumer.py index 85c09bf1a308..1f2b6db17728 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/consumer.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/consumer.py @@ -197,6 +197,7 @@ def receive(self, max_batch_size=None, timeout=None): :rtype: list[~azure.eventhub.common.EventData] :raises: ~azure.eventhub.AuthenticationError, ~azure.eventhub.ConnectError, ~azure.eventhub.ConnectionLostError, ~azure.eventhub.EventHubError + Example: .. literalinclude:: ../examples/test_examples_eventhub.py :start-after: [START eventhub_client_sync_receive] diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py index 6cfd731de589..85020257df46 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py @@ -21,16 +21,42 @@ class EventProcessor(object): + """ + An EventProcessor constantly receives events from all partitions of the Event Hub in the context of a given + consumer group. The received data will be sent to PartitionProcessor to be processed. + + It provides the user a convenient way to receive events from multiple partitions and save checkpoints. + If multiple EventProcessors are running for an event hub, they will automatically balance load. + This load balancing won't be available until preview 3. + + Example: + .. code-block:: python + + class MyPartitionProcessor(PartitionProcessor): + async def process_events(self, events): + if events: + # do something sync or async to process the events + await self._checkpoint_manager.update_checkpoint(events[-1].offset, events[-1].sequence_number) + + import asyncio + from azure.eventhub.aio import EventHubClient + from azure.eventhub.eventprocessor import EventProcessor, PartitionProcessor, Sqlite3PartitionManager + client = EventHubClient.from_connection_string("", receive_timeout=5, retry_total=3) + partition_manager = Sqlite3PartitionManager() + try: + event_processor = EventProcessor(client, "$default", MyPartitionProcessor, partition_manager) + asyncio.ensure_future(event_processor.start()) + await asyncio.sleep(100) # allow it to run 100 seconds + await event_processor.stop() + finally: + await partition_manager.close() + + """ def __init__(self, eventhub_client: EventHubClient, consumer_group_name: str, partition_processor_factory: Callable[[CheckpointManager], PartitionProcessor], partition_manager: PartitionManager, **kwargs): """ - An EventProcessor constantly receives events from all partitions of the Event Hub in the context of a given - consumer group. The received data will be sent to PartitionProcessor to be processed. - - It provides the user a convenient way to receive events from multiple partitions and save checkpoints. - If multiple EventProcessors are running for an event hub, they will automatically balance load. - This load balancing won't be available until preview 3. + Instantiate an EventProcessor. :param eventhub_client: An instance of ~azure.eventhub.aio.EventClient object :type eventhub_client: ~azure.eventhub.aio.EventClient @@ -46,29 +72,6 @@ def __init__(self, eventhub_client: EventHubClient, consumer_group_name: str, :param initial_event_position: The offset to start a partition consumer if the partition has no checkpoint yet. :type initial_event_position: int or str - Example: - ```python - class MyPartitionProcessor(PartitionProcessor): - async def process_events(self, events): - if events: - # do something sync or async to process the events - await self._checkpoint_manager.update_checkpoint(events[-1].offset, events[-1].sequence_number) - - - import asyncio - from azure.eventhub.aio import EventHubClient - from azure.eventhub.eventprocessor import EventProcessor, PartitionProcessor, Sqlite3PartitionManager - client = EventHubClient.from_connection_string("", receive_timeout=5, retry_total=3) - partition_manager = Sqlite3PartitionManager() - try: - event_processor = EventProcessor(client, "$default", MyPartitionProcessor, partition_manager) - asyncio.ensure_future(event_processor.start()) - await asyncio.sleep(100) # allow it to run 100 seconds - await event_processor.stop() - finally: - await partition_manager.close() - ``` - """ self._consumer_group_name = consumer_group_name self._eventhub_client = eventhub_client @@ -87,12 +90,12 @@ def __repr__(self): async def start(self): """Start the EventProcessor. - 1. retrieve the partition ids from eventhubs - 2. claim partition ownership of these partitions. - 3. repeatedly call EvenHubConsumer.receive() to retrieve events and - call user defined PartitionProcessor.process_events() + 1. retrieve the partition ids from eventhubs. + 2. claim partition ownership of these partitions. + 3. repeatedly call EvenHubConsumer.receive() to retrieve events and call user defined PartitionProcessor.process_events(). + + :return: None - :return None """ log.info("EventProcessor %r is being started", self._id) partition_ids = await self._eventhub_client.get_partition_ids() @@ -104,7 +107,8 @@ async def stop(self): This method cancels tasks that are running EventHubConsumer.receive() for the partitions owned by this EventProcessor. - :return None + :return: None + """ for i in range(len(self._tasks)): task = self._tasks.pop() diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py index a06bf48ae084..10aafc79c492 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py @@ -30,10 +30,10 @@ async def close(self, reason): """Called when EventProcessor stops processing this PartitionProcessor. There are different reasons to trigger the PartitionProcessor to close. - Refer to enum class CloseReason + Refer to enum class ~azure.eventhub.eventprocessor.CloseReason :param reason: Reason for closing the PartitionProcessor. - :type reason: CloseReason + :type reason: ~azure.eventhub.eventprocessor.CloseReason """ pass diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py index 13ef7744772b..8655b16bd451 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py @@ -163,6 +163,7 @@ def create_batch(self, max_size=None, partition_key=None): """ Create an EventDataBatch object with max size being max_size. The max_size should be no greater than the max allowed message size defined by the service side. + :param max_size: The maximum size of bytes data that an EventDataBatch object can hold. :type max_size: int :param partition_key: With the given partition_key, event data will land to @@ -204,7 +205,8 @@ def send(self, event_data, partition_key=None, timeout=None): :type partition_key: str :param timeout: The maximum wait time to send the event data. If not specified, the default wait time specified when the producer was created will be used. - :type timeout:float + :type timeout: float + :raises: ~azure.eventhub.AuthenticationError, ~azure.eventhub.ConnectError, ~azure.eventhub.ConnectionLostError, ~azure.eventhub.EventDataError, ~azure.eventhub.EventDataSendError, ~azure.eventhub.EventHubError :return: None diff --git a/sdk/eventhub/azure-eventhubs/examples/test_examples_eventhub.py b/sdk/eventhub/azure-eventhubs/examples/test_examples_eventhub.py index 04e29d08256c..cf8943ef1e6a 100644 --- a/sdk/eventhub/azure-eventhubs/examples/test_examples_eventhub.py +++ b/sdk/eventhub/azure-eventhubs/examples/test_examples_eventhub.py @@ -57,7 +57,7 @@ def test_example_eventhub_sync_send_and_receive(live_eventhub_config): from azure.eventhub import EventData, EventPosition - # [START create_eventhub_client_producer] + # [START create_eventhub_client_sender] client = EventHubClient.from_connection_string(connection_str) # Create a producer. producer = client.create_producer(partition_id="0")