Updating docstings, docs, samples (#6673)

* Draft for updating documentations

* Small improvement

* Updating docstrings, docs and sample

Author: yunhaoling

sdk/eventhub/azure-eventhubs/HISTORY.md

@@ -4,16 +4,22 @@
 
 **New features**
 
-- Added ability to create and send EventDataBatch object with limited data size.
+- Added new class `EventDataBatch` for publication of a batch of events with known size constraint.
+- Added new method `create_batch` to producer for creating EventDataBatch objects.
 - Added new configuration parameters for exponential delay among each retry operation.
     - `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`.
 
 **Breaking changes**
 
-- New `EventProcessor` design
-    - The `EventProcessorHost` was waived.
+- Replaced `max_retries` configuration parameter of the EventHubClient with `retry_total`.
+- Introduced the initial concept of a new version of the `EventProcessor`, intended as a neutral framework for processing events across all partitions for a given Event Hub and in the context of a specific Consumer Group. This early preview is intended to allow consumers to test the new design using a single instance that does not persist checkpoints to any durable store.
+    - `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.
 
 ## 5.0.0b1 (2019-06-25)
 

sdk/eventhub/azure-eventhubs/README.md

@@ -90,6 +90,7 @@ The following sections provide several code snippets covering some of the most c
 - [Consume events from an Event Hub](#consume-events-from-an-event-hub)
 - [Async publish events to an Event Hub](#async-publish-events-to-an-event-hub)
 - [Async consume events from an Event Hub](#async-consume-events-from-an-event-hub)
+- [Consume events from all partitions of an Event Hub](#consume-events-from-all-partitions-of-an-event-hub)
 
 ### Inspect an Event Hub
 
@@ -206,6 +207,54 @@ finally:
     pass
 ```
 
+### Consume events from all partitions of an Event Hub
+
+To consume events from all partitions of an Event Hub, you'll create an `EventProcessor` for a specific consumer group. When an Event Hub is created, it provides a default consumer group that can be used to get started.
+
+The `EventProcessor` will delegate processing of events to a `PartitionProcessor` implementation that you provide, allowing your logic to focus on the logic needed to provide value while the processor holds responsibility for managing the underlying consumer operations. In our example, we will focus on building the `EventProcessor` and use a very minimal partition processor that does no actual processing.
+
+```python
+import asyncio
+
+from azure.eventhub.aio import EventHubClient
+from azure.eventhub.eventprocessor import EventProcessor, PartitionProcessor, Sqlite3PartitionManager
+
+connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
+
+async def do_operation(event):
+    # do some sync or async operations. If the operation is i/o intensive, async will have better performance
+    print(event)
+
+class MyPartitionProcessor(PartitionProcessor):
+    def __init__(self, checkpoint_manager):
+        super(MyPartitionProcessor, self).__init__(checkpoint_manager)
+
+    async def process_events(self, events):
+        if events:
+            await asyncio.gather(*[do_operation(event) for event in events])
+            await self._checkpoint_manager.update_checkpoint(events[-1].offset, events[-1].sequence_number)
+
+def partition_processor_factory(checkpoint_manager):
+    return MyPartitionProcessor(checkpoint_manager)
+
+async def main():
+    client = EventHubClient.from_connection_string(connection_str, receive_timeout=5, retry_total=3)
+    partition_manager = Sqlite3PartitionManager()
+    try:
+        event_processor = EventProcessor(client, "$default", MyPartitionProcessor, partition_manager)
+        # You can also define a callable object for creating PartitionProcessor like below:
+        # event_processor = EventProcessor(client, "$default", partition_processor_factory, partition_manager)
+        asyncio.ensure_future(event_processor.start())
+        await asyncio.sleep(60)
+        await event_processor.stop()
+    finally:
+        await partition_manager.close()
+
+if __name__ == '__main__':
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(main())
+```
+
 ## Troubleshooting
 
 ### General
@@ -230,7 +279,7 @@ These are the samples in our repo demonstraing the usage of the library.
 - [./examples/recv.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/recv.py) - use consumer to consume events
 - [./examples/async_examples/send_async.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/async_examples/send_async.py) - async/await support of a producer
 - [./examples/async_examples/recv_async.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/async_examples/recv_async.py) - async/await support of a consumer
-- [./examples/eph.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/eph.py) - event processor host
+- [./examples/eventprocessor/event_processor_example.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/eventprocessor/event_processor_example.py) - event processor
 
 ### Documentation
 

sdk/eventhub/azure-eventhubs/azure/eventhub/_consumer_producer_mixin.py

@@ -15,9 +15,9 @@
 
 def _retry_decorator(to_be_wrapped_func):
     def wrapped_func(self, *args, **kwargs):
-        timeout = kwargs.pop("timeout", None)
+        timeout = kwargs.pop("timeout", 100000)
         if not timeout:
-            timeout = 100000  # timeout None or 0 mean no timeout. 100000 seconds is equivalent to no timeout
+            timeout = 100000  # timeout equals to 0 means no timeout, set the value to be a large number.
         timeout_time = time.time() + timeout
         max_retries = self.client.config.max_retries
         retry_count = 0

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/_consumer_producer_mixin_async.py

@@ -15,9 +15,9 @@
 
 def _retry_decorator(to_be_wrapped_func):
     async def wrapped_func(self, *args, **kwargs):
-        timeout = kwargs.pop("timeout", None)
+        timeout = kwargs.pop("timeout", 100000)
         if not timeout:
-            timeout = 100000  # timeout None or 0 mean no timeout. 100000 seconds is equivalent to no timeout
+            timeout = 100000  # timeout equals to 0 means no timeout, set the value to be a large number.
         timeout_time = time.time() + timeout
         max_retries = self.client.config.max_retries
         retry_count = 0

sdk/eventhub/azure-eventhubs/azure/eventhub/client_abstract.py

@@ -118,9 +118,9 @@ def __init__(self, host, event_hub_path, credential, **kwargs):
         :type auth_timeout: float
         :param user_agent: The user agent that needs to be appended to the built in user agent string.
         :type user_agent: str
-        :param max_retries: The max number of attempts to redo the failed operation when an error happened. Default
+        :param retry_total: The total number of attempts to redo the failed operation when an error happened. Default
          value is 3.
-        :type max_retries: int
+        :type retry_total: int
         :param transport_type: The type of transport protocol that will be used for communicating with
          the Event Hubs service. Default is ~azure.eventhub.TransportType.Amqp.
         :type transport_type: ~azure.eventhub.TransportType
@@ -239,9 +239,9 @@ def from_connection_string(cls, conn_str, **kwargs):
         :type auth_timeout: float
         :param user_agent: The user agent that needs to be appended to the built in user agent string.
         :type user_agent: str
-        :param max_retries: The max number of attempts to redo the failed operation when an error happened. Default
+        :param retry_total: The total number of attempts to redo the failed operation when an error happened. Default
          value is 3.
-        :type max_retries: int
+        :type retry_total: int
         :param transport_type: The type of transport protocol that will be used for communicating with
          the Event Hubs service. Default is ~azure.eventhub.TransportType.Amqp.
         :type transport_type: ~azure.eventhub.TransportType

sdk/eventhub/azure-eventhubs/azure/eventhub/common.py

@@ -63,8 +63,6 @@ def __init__(self, body=None, to_device=None):
 
         :param body: The data to send in a single message.
         :type body: str, bytes or list
-        :param batch: A data generator to send batched messages.
-        :type batch: Generator
         :param to_device: An IoT device to route to.
         :type to_device: str
         """

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/checkpoint_manager.py

@@ -8,7 +8,9 @@
 
 
 class CheckpointManager(object):
-    """Every PartitionProcessor has a CheckpointManager to save the partition's checkpoint.
+    """
+    CheckpointManager is responsible for the creation of checkpoints.
+    The interaction with the chosen storage service is done via ~azure.eventhub.eventprocessor.PartitionManager.
 
     """
     def __init__(self, partition_id: str, eventhub_name: str, consumer_group_name: str, owner_id: str, partition_manager: PartitionManager):
@@ -19,10 +21,13 @@ def __init__(self, partition_id: str, eventhub_name: str, consumer_group_name: s
         self.partition_manager = partition_manager
 
     async def update_checkpoint(self, offset, sequence_number=None):
-        """Users call this method in PartitionProcessor.process_events() to save checkpoints
+        """
+        Updates the checkpoint using the given information for the associated partition and consumer group in the chosen storage service.
 
-        :param offset: offset of the processed EventData
-        :param sequence_number: sequence_number of the processed EventData
+        :param offset: The offset of the ~azure.eventhub.EventData the new checkpoint will be associated with.
+        :type offset: str
+        :param sequence_number: The sequence_number of the ~azure.eventhub.EventData the new checkpoint will be associated with.
+        :type sequence_number: int
         :return: None
         """
         await self.partition_manager.update_checkpoint(

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py

@@ -24,23 +24,26 @@ class EventProcessor(object):
     def __init__(self, eventhub_client: EventHubClient, consumer_group_name: str,
                  partition_processor_factory: Callable[[CheckpointManager], PartitionProcessor],
                  partition_manager: PartitionManager, **kwargs):
-        """An EventProcessor automatically creates and runs consumers for all partitions of the eventhub.
+        """
+        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 loading.
-        This load balancer won't be available until preview 3.
-
-        :param eventhub_client: an instance of azure.eventhub.aio.EventClient object
-        :param consumer_group_name: the consumer group that is used to receive events from
-        :param partition_processor_factory: a callable (type or function) that is called to return a PartitionProcessor.
-        Users define their own PartitionProcessor by subclassing it implement abstract method process_events() to
-        implement their own operation logic.
-        :param partition_manager: an instance of a PartitionManager implementation. A partition manager claims ownership
-        of partitions and saves partition checkpoints to a data storage. For preview 2, sample Sqlite3PartitionManager is
-        already provided. For the future releases there will be more PartitionManager implementation provided to save
-        checkpoint to different data storage. Users can also implement their PartitionManager class to user their own
-        preferred data storage.
-        :param initial_event_position: the offset to start a partition consumer if the partition has no checkpoint yet
+        If multiple EventProcessors are running for an event hub, they will automatically balance load.
+        This load balancing won't be available until preview 3.
+
+        :param eventhub_client: An instance of ~azure.eventhub.aio.EventClient object
+        :type eventhub_client: ~azure.eventhub.aio.EventClient
+        :param consumer_group_name: The name of the consumer group this event processor is associated with. Events will
+         be read only in the context of this group.
+        :type consumer_group_name: str
+        :param partition_processor_factory: A callable(type or function) object that creates an instance of a class
+         implementing the ~azure.eventhub.eventprocessor.PartitionProcessor interface.
+        :type partition_processor_factory: callable object
+        :param partition_manager: Interacts with the storage system, dealing with ownership and checkpoints.
+         For preview 2, sample Sqlite3PartitionManager is provided.
+        :type partition_manager: Class implementing the ~azure.eventhub.eventprocessor.PartitionManager interface.
+        :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:

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_manager.py

@@ -8,16 +8,21 @@
 
 
 class PartitionManager(ABC):
-    """Subclass PartitionManager to implement the read/write access to storage service to list/claim ownership and save checkpoint.
-
+    """
+    PartitionManager deals with the interaction with the chosen storage service.
+    It's able to list/claim ownership and create checkpoint.
     """
 
     @abstractmethod
     async def list_ownership(self, eventhub_name: str, consumer_group_name: str) -> Iterable[Dict[str, Any]]:
         """
+        Retrieves a complete ownership list from the chosen storage service.
 
-        :param eventhub_name:
-        :param consumer_group_name:
+        :param eventhub_name: The name of the specific Event Hub the ownership are associated with, relative to
+         the Event Hubs namespace that contains it.
+        :type eventhub_name: str
+        :param consumer_group_name: The name of the consumer group the ownership are associated with.
+        :type consumer_group_name: str
         :return: Iterable of dictionaries containing the following partition ownership information:
                 eventhub_name
                 consumer_group_name
@@ -33,11 +38,45 @@ async def list_ownership(self, eventhub_name: str, consumer_group_name: str) ->
 
     @abstractmethod
     async def claim_ownership(self, partitions: Iterable[Dict[str, Any]]) -> Iterable[Dict[str, Any]]:
+        """
+        Tries to claim a list of specified ownership.
+
+        :param partitions: Iterable of dictionaries containing all the ownership to claim.
+        :type partitions: Iterable of dict
+        :return: Iterable of dictionaries containing the following partition ownership information:
+                eventhub_name
+                consumer_group_name
+                owner_id
+                partition_id
+                owner_level
+                offset
+                sequence_number
+                last_modified_time
+                etag
+        """
         pass
 
     @abstractmethod
     async def update_checkpoint(self, eventhub_name, consumer_group_name, partition_id, owner_id,
                                 offset, sequence_number) -> None:
+        """
+        Updates the checkpoint using the given information for the associated partition and consumer group in the chosen storage service.
+
+        :param eventhub_name: The name of the specific Event Hub the ownership are associated with, relative to
+         the Event Hubs namespace that contains it.
+        :type eventhub_name: str
+        :param consumer_group_name: The name of the consumer group the ownership are associated with.
+        :type consumer_group_name: str
+        :param partition_id: The partition id which the checkpoint is created for.
+        :type partition_id: str
+        :param owner_id: The identifier of the ~azure.eventhub.eventprocessor.EventProcessor.
+        :type owner_id: str
+        :param offset: The offset of the ~azure.eventhub.EventData the new checkpoint will be associated with.
+        :type offset: str
+        :param sequence_number: The sequence_number of the ~azure.eventhub.EventData the new checkpoint will be associated with.
+        :type sequence_number: int
+        :return:
+        """
         pass
 
     async def close(self):

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py

@@ -18,6 +18,11 @@ class CloseReason(Enum):
 
 
 class PartitionProcessor(ABC):
+    """
+    PartitionProcessor processes events received from the Azure Event Hubs service. A single instance of a class
+    implementing this interface will be created for every partition the associated ~azure.eventhub.eventprocessor.EventProcessor owns.
+
+    """
     def __init__(self, checkpoint_manager: CheckpointManager):
         self._checkpoint_manager = checkpoint_manager
 
@@ -27,18 +32,27 @@ async def close(self, reason):
         There are different reasons to trigger the PartitionProcessor to close.
         Refer to enum class CloseReason
 
+        :param reason: Reason for closing the PartitionProcessor.
+        :type reason: CloseReason
+
         """
         pass
 
     @abstractmethod
     async def process_events(self, events: List[EventData]):
         """Called when a batch of events have been received.
 
+        :param events: Received events.
+        :type events: list[~azure.eventhub.common.EventData]
+
         """
         pass
 
     async def process_error(self, error):
         """Called when an error happens
 
+        :param error: The error that happens.
+        :type error: Exception
+
         """
         pass