Polish async retries and uploads

Author: kvz

tests/test_async_client.py

@@ -392,7 +392,7 @@ async def test_async_client_delete_template_get_bill_and_plain_text_fallback(sel
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.headers["X-Async-Route"], "get_assembly_plain")
 
-    async def test_async_assembly_create_returns_plain_text_response_without_crashing(self):
+    async def test_async_assembly_create_raises_on_plain_text_error_response(self):
         plain_response = Response(
             data="plain assembly response",
             status_code=502,
@@ -403,17 +403,12 @@ async def test_async_assembly_create_returns_plain_text_response_without_crashin
             assembly = client.new_assembly()
 
             with mock.patch.object(client.request, "post", new=mock.AsyncMock(return_value=plain_response)) as post_mock:
-                with mock.patch.object(client, "get_assembly", new=mock.AsyncMock()) as get_mock:
-                    with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock) as sleep_mock:
-                        response = await assembly.create(wait=True, resumable=False)
+                with self.assertRaises(RuntimeError):
+                    await assembly.create(wait=True, resumable=False)
 
-        self.assertIs(response, plain_response)
-        self.assertEqual(response.data, "plain assembly response")
         post_mock.assert_awaited_once()
-        get_mock.assert_not_awaited()
-        sleep_mock.assert_not_awaited()
 
-    async def test_async_assembly_wait_returns_plain_text_poll_response_without_crashing(self):
+    async def test_async_assembly_wait_raises_on_plain_text_poll_response(self):
         initial_response = Response(
             data={
                 "ok": "ASSEMBLY_PROCESSING",
@@ -435,10 +430,9 @@ async def test_async_assembly_wait_returns_plain_text_poll_response_without_cras
             with mock.patch.object(client.request, "post", new=mock.AsyncMock(return_value=initial_response)) as post_mock:
                 with mock.patch.object(client, "get_assembly", new=mock.AsyncMock(return_value=plain_response)) as get_mock:
                     with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock) as sleep_mock:
-                        response = await assembly.create(wait=True, resumable=False)
+                        with self.assertRaises(RuntimeError):
+                            await assembly.create(wait=True, resumable=False)
 
-        self.assertIs(response, plain_response)
-        self.assertEqual(response.data, "plain assembly response")
         post_mock.assert_awaited_once()
         get_mock.assert_awaited_once_with(
             assembly_url=f"{self.server.base_url}/assemblies/assembly-123"
@@ -988,10 +982,18 @@ async def test_async_assembly_wait_returns_last_poll_response_when_budget_exhaus
 
         self.assertEqual(response.data["error"], "RATE_LIMIT_REACHED")
         post_mock.assert_awaited_once()
-        get_mock.assert_awaited_once_with(
-            assembly_url=f"{self.server.base_url}/assemblies/assembly-123"
+        self.assertEqual(
+            get_mock.await_args_list,
+            [
+                mock.call(
+                    assembly_url=f"{self.server.base_url}/assemblies/assembly-123"
+                ),
+                mock.call(
+                    assembly_url=f"{self.server.base_url}/assemblies/assembly-123"
+                ),
+            ],
         )
-        self.assertEqual(sleep_mock.await_args_list, [mock.call(0)])
+        self.assertEqual(sleep_mock.await_args_list, [mock.call(0), mock.call(0)])
 
     async def test_async_assembly_non_resumable_rate_limit_rewinds_files_for_retry(self):
         reads = []
@@ -1066,6 +1068,25 @@ async def test_async_request_uses_connect_and_read_timeouts_for_uploads(self):
         self.assertIsNone(timeout.sock_read)
         self.assertEqual(session.calls[0][1]["data"]._fields[2][1]["Content-Type"], "image/jpeg")
 
+    async def test_async_request_filters_none_and_lowercases_booleans_in_extra_data(self):
+        session = _RecordingSession({"ok": "ASSEMBLY_COMPLETED"})
+        client = AsyncTransloadit("key", "secret", service=self.server.base_url, session=session)
+        upload = io.BytesIO(b"payload")
+        upload.name = "clip.jpg"
+
+        response = await client.request.post(
+            "/assemblies",
+            data={"foo": "bar"},
+            extra_data={"enabled": True, "skip": None},
+            files={"file": upload},
+        )
+
+        self.assertEqual(response.data["ok"], "ASSEMBLY_COMPLETED")
+        fields = {field[0]["name"]: field for field in session.calls[0][1]["data"]._fields}
+        self.assertIn("enabled", fields)
+        self.assertNotIn("skip", fields)
+        self.assertEqual(fields["enabled"][2], "true")
+
     async def test_async_request_uses_filename_fallback_for_trailing_slash_stream_name(self):
         session = _RecordingSession({"ok": "ASSEMBLY_COMPLETED"})
         client = AsyncTransloadit("key", "secret", service=self.server.base_url, session=session)

transloadit/async_assembly.py

@@ -101,6 +101,8 @@ async def create(self, wait=False, resumable=True, retries=3):
 
             response_data = self._response_data(response)
             if response_data is None:
+                if response.status_code >= 400:
+                    raise RuntimeError(f"Unexpected non-JSON response ({response.status_code}).")
                 return response
 
             if self._rate_limit_reached(response_data):
@@ -130,20 +132,23 @@ async def create(self, wait=False, resumable=True, retries=3):
 
                 poll_response = response
                 poll_data = response_data
-                remaining_polls = poll_retries
+                remaining_rate_limit_retries = poll_retries
                 while not self._assembly_finished(poll_data):
-                    if remaining_polls <= 0:
-                        return poll_response
+                    if self._rate_limit_reached(poll_data):
+                        if remaining_rate_limit_retries <= 0:
+                            return poll_response
+                        remaining_rate_limit_retries -= 1
                     sleep_time = poll_data.get("info", {}).get("retryIn", 1)
                     await asyncio.sleep(sleep_time)
                     poll_response = await self.transloadit.get_assembly(
                         assembly_url=assembly_url or poll_data.get("assembly_ssl_url")
                     )
                     poll_data = self._response_data(poll_response)
                     if poll_data is None:
+                        if poll_response.status_code >= 400:
+                            raise RuntimeError(f"Unexpected non-JSON response ({poll_response.status_code}).")
                         return poll_response
                     assembly_url = poll_data.get("assembly_ssl_url") or assembly_url
-                    remaining_polls -= 1
 
                 return poll_response
 

transloadit/async_request.py

@@ -41,6 +41,9 @@ def __init__(self, transloadit, session=None):
     def session(self):
         return self._session
 
+    def _headers(self):
+        return dict(self.HEADERS)
+
     async def _ensure_session(self):
         async with self._session_lock:
             if self._session is None:
@@ -66,7 +69,15 @@ def _timeout(self, files=False):
         )
 
     def _normalize_payload(self, data):
-        return {key: str(value) for key, value in data.items()}
+        normalized = {}
+        for key, value in data.items():
+            if value is None:
+                continue
+            if isinstance(value, bool):
+                normalized[key] = "true" if value else "false"
+            else:
+                normalized[key] = str(value)
+        return normalized
 
     async def _read_response_data(self, response):
         try:
@@ -82,7 +93,7 @@ async def get(self, path, params=None):
         async with session.get(
             self._get_full_url(path),
             params=self._to_payload(params),
-            headers=self.HEADERS,
+            headers=self._headers(),
             timeout=self._timeout(),
         ) as response:
             return Response(
@@ -116,7 +127,7 @@ async def post(self, path, data=None, extra_data=None, files=None):
         async with session.post(
             self._get_full_url(path),
             data=payload,
-            headers=self.HEADERS,
+            headers=self._headers(),
             timeout=self._timeout(files=bool(files)),
         ) as response:
             return Response(
@@ -134,7 +145,7 @@ async def put(self, path, data=None):
         async with session.put(
             self._get_full_url(path),
             data=data,
-            headers=self.HEADERS,
+            headers=self._headers(),
             timeout=self._timeout(),
         ) as response:
             return Response(
@@ -152,7 +163,7 @@ async def delete(self, path, data=None):
         async with session.delete(
             self._get_full_url(path),
             data=data,
-            headers=self.HEADERS,
+            headers=self._headers(),
             timeout=self._timeout(),
         ) as response:
             return Response(