improved docstrings

Author: daniel-sanche

google/api_core/retry.py

@@ -311,6 +311,9 @@ class Retry(object):
             will be wrapped with retry logic, and any failed outputs will
             restart the stream. If False, only the input function call itself
             will be retried. Defaults to False.
+            To avoid duplicate values, retryable streams should typically be
+            wrapped in additional filter logic before use. For more details, see
+            ``google/api_core/retry_streaming.RetryaleGenerator``.
         deadline (float): DEPRECATED: use `timeout` instead. For backward
             compatibility, if specified it will override the ``timeout`` parameter.
     """

google/api_core/retry_async.py

@@ -173,8 +173,6 @@ class AsyncRetry:
         maximum (float): The maximum amount of time to delay in seconds.
         multiplier (float): The multiplier applied to the delay.
         timeout (float): How long to keep retrying in seconds.
-            When ``is_stream``, only time spent waiting on the
-            target or sleeping between retries is counted towards the timeout.
         on_error (Callable[Exception]): A function to call while processing
             a retryable exception. Any error raised by this function will
             *not* be caught.
@@ -184,6 +182,9 @@ class AsyncRetry:
             If True, the iterable will be wrapped with retry logic, and any
             failed outputs will restart the stream. If False, only the input
             function call itself will be retried. Defaults to False.
+            To avoid duplicate values, retryable streams should typically be
+            wrapped in additional filter logic before use. For more details, see
+            ``google.api_core.retry_streaming_async.AsyncRetryableGenerator``.
         deadline (float): DEPRECATED use ``timeout`` instead. If set it will
         override ``timeout`` parameter.
     """

google/api_core/retry_streaming.py

@@ -30,8 +30,72 @@
 
 class RetryableGenerator(Generator[T, Any, None]):
     """
-    Helper class for retrying Iterator and Generator-based
-    streaming APIs.
+    Generator wrapper for retryable streaming RPCs.
+    RetryableGenerator will be used when initilizing a retry with
+    ``Retry(is_stream=True)``.
+
+    When ``is_stream=False``, the target is treated as a callable,
+    and will retry when the callable returns an error. When ``is_stream=True``,
+    the target will be treated as a callable that retruns an iterable. Instead
+    of just wrapping the initial call in retry logic, the entire iterable is
+    wrapped, with each yield passing through RetryableGenerator. If any yield
+    in the stream raises a retryable exception, the entire stream will be
+    retried.
+
+    Important Note: when a stream is encounters a retryable error, it will
+    silently construct a fresh iterator instance in the background
+    and continue yielding (likely duplicate) values as if no error occurred.
+    This is the most general way to retry a stream, but it often is not the
+    desired behavior. Example: iter([1, 2, 1/0]) -> [1, 2, 1, 2, ...]
+
+    There are two ways to build more advanced retry logic for streams:
+
+    1. Wrap the target
+        Use a ``target`` that maintains state between retries, and creates a 
+        different generator on each retry call. For example, you can wrap a 
+        network call in a function that modifies the request based on what has 
+        already been returned:
+
+        ```
+        def attempt_with_modified_request(target, request, seen_items=[]):
+            # remove seen items from request on each attempt
+            new_request = modify_request(request, seen_items)
+            new_generator = target(new_request)
+            for item in new_generator:
+                yield item
+                seen_items.append(item)
+
+        retry_wrapped = Retry(is_stream=True)(attempt_with_modified_request, target, request, [])
+        ```
+
+        2. Wrap the RetryableGenerator
+            Alternatively, you can wrap the RetryableGenerator itself before
+            passing it to the end-user to add a filter on the stream. For
+            example, you can keep track of the items that were successfully yielded
+            in previous retry attempts, and only yield new items when the
+            new attempt surpasses the previous ones:
+
+            ``
+            def retryable_with_filter(target):
+                stream_idx = 0
+                # reset stream_idx when the stream is retried
+                def on_error(e):
+                    nonlocal stream_idx
+                    stream_idx = 0
+                # build retryable
+                retryable_gen = Retry(is_stream=True, on_error=on_error, ...)(target)
+                # keep track of what has been yielded out of filter
+                yielded_items = []
+                for item in retryable_gen:
+                    if stream_idx >= len(yielded_items):
+                        yield item
+                        yielded_items.append(item)
+                    elif item != previous_stream[stream_idx]:
+                        raise ValueError("Stream differs from last attempt")"
+                    stream_idx += 1
+
+            filter_retry_wrapped = retryable_with_filter(target)
+            ```
     """
 
     def __init__(

google/api_core/retry_streaming_async.py

@@ -41,8 +41,72 @@
 
 class AsyncRetryableGenerator(AsyncGenerator[T, None]):
     """
-    Helper class for retrying AsyncIterator and AsyncGenerator-based
-    streaming APIs.
+    AsyncGenerator wrapper for retryable streaming RPCs.
+    AsyncRetryableGenerator will be used when initilizing a retry with
+    ``AsyncRetry(is_stream=True)``.
+
+    When ``is_stream=False``, the target is treated as a coroutine,
+    and will retry when the coroutine returns an error. When ``is_stream=True``,
+    the target will be treated as a callable that retruns an AsyncIterable. Instead
+    of just wrapping the initial call in retry logic, the entire iterable is
+    wrapped, with each yield passing through AsyncRetryableGenerator. If any yield
+    in the stream raises a retryable exception, the entire stream will be
+    retried.
+
+    Important Note: when a stream is encounters a retryable error, it will
+    silently construct a fresh iterator instance in the background
+    and continue yielding (likely duplicate) values as if no error occurred.
+    This is the most general way to retry a stream, but it often is not the
+    desired behavior. Example: iter([1, 2, 1/0]) -> [1, 2, 1, 2, ...]
+
+    There are two ways to build more advanced retry logic for streams:
+
+    1. Wrap the target
+        Use a ``target`` that maintains state between retries, and creates a
+        different generator on each retry call. For example, you can wrap a
+        grpc call in a function that modifies the request based on what has
+        already been returned:
+
+        ```
+        async def attempt_with_modified_request(target, request, seen_items=[]):
+            # remove seen items from request on each attempt
+            new_request = modify_request(request, seen_items)
+            new_generator = await target(new_request)
+            async for item in new_generator:
+                yield item
+                seen_items.append(item)
+
+        retry_wrapped = AsyncRetry(is_stream=True)(attempt_with_modified_request, target, request, [])
+        ```
+
+        2. Wrap the AsyncRetryableGenerator
+            Alternatively, you can wrap the AsyncRetryableGenerator itself before
+            passing it to the end-user to add a filter on the stream. For
+            example, you can keep track of the items that were successfully yielded
+            in previous retry attempts, and only yield new items when the
+            new attempt surpasses the previous ones:
+
+            ``
+            async def retryable_with_filter(target):
+                stream_idx = 0
+                # reset stream_idx when the stream is retried
+                def on_error(e):
+                    nonlocal stream_idx
+                    stream_idx = 0
+                # build retryable
+                retryable_gen = AsyncRetry(is_stream=True, on_error=on_error, ...)(target)
+                # keep track of what has been yielded out of filter
+                yielded_items = []
+                async for item in retryable_gen:
+                    if stream_idx >= len(yielded_items):
+                        yield item
+                        yielded_items.append(item)
+                    elif item != previous_stream[stream_idx]:
+                        raise ValueError("Stream differs from last attempt")"
+                    stream_idx += 1
+
+            filter_retry_wrapped = retryable_with_filter(target)
+            ```
     """
 
     def __init__(