estuary-cdk: retry connection timeouts

It's a good idea to retry connection timeouts a few times in case of
intermittent errors from flaky APIs. This is safe to do in the request()
method since it consumes the entire response body before returning,
unlike request_lines() and request_stream() which yield chunks that may
have already been processed before a timeout occurs.

Author: Alex-Bair

estuary-cdk/estuary_cdk/http.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from logging import Logger
+from aiohttp.client_exceptions import ConnectionTimeoutError
 from estuary_cdk.incremental_json_processor import Remainder
 from pydantic import BaseModel
 from typing import AsyncGenerator, Any, TypeVar, Union, Callable
@@ -78,20 +79,34 @@ async def request(
     ) -> bytes:
         """Request a url and return its body as bytes"""
 
-        chunks: list[bytes] = []
-        _, body_generator = await self._request_stream(
-            log, url, method, params, json, form, _with_token, headers
-        )
+        max_attempts = 3
+        attempt = 1
 
-        async for chunk in body_generator():
-            chunks.append(chunk)
+        while True:
+            try:
+                chunks: list[bytes] = []
+                _, body_generator = await self._request_stream(
+                    log, url, method, params, json, form, _with_token, headers
+                )
 
-        if len(chunks) == 0:
-            return b""
-        elif len(chunks) == 1:
-            return chunks[0]
-        else:
-            return b"".join(chunks)
+                async for chunk in body_generator():
+                    chunks.append(chunk)
+
+                if len(chunks) == 0:
+                    return b""
+                elif len(chunks) == 1:
+                    return chunks[0]
+                else:
+                    return b"".join(chunks)
+            except ConnectionTimeoutError as e:
+                if attempt <= max_attempts:
+                    log.warning(
+                        f"Connection timeout error (will retry)",
+                        {"url": url, "method": method, "attempt": attempt, "error": str(e)}
+                    )
+                    attempt += 1
+                else:
+                    raise
 
     async def request_lines(
         self,