Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,8 @@
## 0.46.4

### Fixes
* Return API errors immediately when their JSON body declares `"retryable": false`, instead of retrying them until the retry budget expires. Applies to sync and async requests, including split-PDF chunks. Other errors keep the existing status-based retry behavior.

## 0.46.3

### Fixes
Expand Down
160 changes: 160 additions & 0 deletions _test_unstructured_client/unit/test_retries.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
import httpx
import pytest

from unstructured_client import UnstructuredClient
from unstructured_client.models.errors import ServerError
from unstructured_client.utils import retries as retries_module
from unstructured_client.utils.retries import (
BackoffStrategy,
Expand Down Expand Up @@ -555,3 +557,161 @@ async def func():
with pytest.raises(ValueError, match="permanent"):
asyncio.run(retry_async(func, retries_config))
assert call_count == 1


def _send_with_retry(respond, config, async_mode, *, stream=False):
transport = httpx.MockTransport(respond)
if async_mode:
async def run():
async with httpx.AsyncClient(transport=transport) as client:
return await retry_async(
lambda: client.send(client.build_request("GET", "http://test.invalid"), stream=stream),
config,
)
return asyncio.run(run())
with httpx.Client(transport=transport) as client:
return retry(
lambda: client.send(client.build_request("GET", "http://test.invalid"), stream=stream),
config,
)


@pytest.mark.parametrize("async_mode", [False, True])
@pytest.mark.parametrize(
"body,content_type,stop",
[
(b'{"detail":"unsupported","retryable":false}', "application/json", True),
(b'{"retryable":false}', "application/problem+json; charset=utf-8", True),
(b'{"retryable":true}', "application/json", False),
(b'{"detail":"Internal server error"}', "application/json", False),
(b'{"retryable":"false"}', "application/json", False),
(b'{"retryable":0}', "application/json", False),
(b'{"retryable":null}', "application/json", False),
(b'[{"retryable":false}]', "application/json", False),
(b'false', "application/json", False),
(b'{broken', "application/json", False),
pytest.param(
b'{"retryable":false,"detail":' + b"[" * 100_000 + b"]" * 100_000 + b"}",
"application/json",
False,
id="deeply-nested",
),
(b'upstream error', "text/plain", False),
(b'{"retryable":false}', "text/plain", True),
(b'{"retryable":false}', "", True),
],
)
def test_declared_retry_verdict(
fake_clock, async_mode, body, content_type, stop
):
status_code = 503
attempts = []

def respond(request):
attempts.append(request)
if len(attempts) == 1:
return httpx.Response(status_code, content=body, headers={"Content-Type": content_type})
return httpx.Response(200, json=[])

config = _retries(status_codes=["5xx"], initial_interval=1, max_interval=1)
result = _send_with_retry(respond, config, async_mode)

assert len(attempts) == (1 if stop else 2)
assert result.status_code == (status_code if stop else 200)
if stop:
assert result.content == body


@pytest.mark.parametrize("async_mode", [False, True])
@pytest.mark.parametrize("verdict", [False, True])
def test_declared_verdict_does_not_retry_non_matching_status(fake_clock, async_mode, verdict):
attempts = []

def respond(request):
attempts.append(request)
return httpx.Response(400, json={"retryable": verdict})

config = _retries(status_codes=["5xx"])
response = _send_with_retry(respond, config, async_mode)

assert response.status_code == 400
assert len(attempts) == 1


@pytest.mark.parametrize("async_mode", [False, True])
def test_unread_stream_is_not_consumed(fake_clock, async_mode):
class SyncStream(httpx.SyncByteStream):
def __iter__(self):
pytest.fail("retry inspected an unread stream")
yield b""

class AsyncStream(httpx.AsyncByteStream):
async def __aiter__(self):
pytest.fail("retry inspected an unread stream")
yield b""

attempts = []

def respond(request):
attempts.append(request)
if len(attempts) == 1:
return httpx.Response(
503, headers={"Content-Type": "application/json"},
stream=AsyncStream() if async_mode else SyncStream(),
)
return httpx.Response(200)

config = _retries(status_codes=["5xx"])
result = _send_with_retry(respond, config, async_mode, stream=True)

assert result.status_code == 200
assert len(attempts) == 2


@pytest.mark.parametrize("async_mode", [False, True])
@pytest.mark.parametrize("status_code", [500, 503])
def test_public_partition_preserves_declared_error(fake_clock, async_mode, status_code):
body = {"detail": "unsupported document", "retryable": False}
attempts = []

def respond(request):
attempts.append(request)
return httpx.Response(status_code, json=body)

async def run():
transport = httpx.MockTransport(respond)
with httpx.Client(transport=transport) as client:
async with httpx.AsyncClient(transport=transport) as async_client:
sdk = UnstructuredClient(
server_url="http://test.invalid", client=client, async_client=async_client
)
request = {"partition_parameters": {
"files": {"content": b"hello", "file_name": "test.txt"},
"split_pdf_page": False,
}}
config = _retries(initial_interval=1, max_interval=1, max_elapsed_time=2).config
with pytest.raises(ServerError) as exc:
if async_mode:
await sdk.general.partition_async(request=request, retries=config)
else:
sdk.general.partition(request=request, retries=config)
assert exc.value.raw_response.json() == body
assert "unsupported document" in str(exc.value)

asyncio.run(run())
assert len(attempts) == 1


@pytest.mark.parametrize("async_mode", [False, True])
@pytest.mark.parametrize("status_code", [500, 503])
@pytest.mark.parametrize("status_codes", [["5xx"], ["500", "503"]])
def test_declared_verdict_stops_matching_status(fake_clock, async_mode, status_code, status_codes):
attempts = []

def respond(request):
attempts.append(request)
return httpx.Response(status_code, json={"retryable": False})

response = _send_with_retry(respond, _retries(status_codes=status_codes), async_mode)
assert response.status_code == status_code
assert len(attempts) == 1
37 changes: 36 additions & 1 deletion _test_unstructured_client/unit/test_split_pdf_hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -2186,4 +2186,39 @@ def _chunk_paths_side_effect(*args, **kwargs):
assert hook.cache_tmp_data_feature == {}
assert hook.cache_tmp_data_dir == {}
tempdir.cleanup.assert_called_once()
executor.shutdown.assert_not_called()
executor.shutdown.assert_not_called()


@pytest.mark.asyncio
@pytest.mark.parametrize("cached", [False, True])
@pytest.mark.parametrize("status_code", [500, 503])
async def test_declared_non_retryable_chunk_sent_once(tmp_path, cached, status_code):
body = {"detail": "unsupported chunk", "retryable": False}
attempts = []

def respond(request):
attempts.append(request)
return httpx.Response(status_code, json=body)

async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
response = await SplitPdfHook().call_api_partial(
pdf_chunk_request=httpx.Request(
"POST",
"http://test.invalid/general/v0/general",
headers={"Accept": "application/x-ndjson" if cached else "application/json"},
content=b"pdf chunk",
),
pdf_chunk_file=io.BytesIO(b"pdf chunk"),
limiter=asyncio.Semaphore(1),
_operation_id="declared-error",
chunk_index=0,
page_number=1,
async_client=client,
retry_config=RetryConfig("backoff", BackoffStrategy(1, 1, 1, 2), True),
cache_tmp_data_feature=cached,
temp_dir_path=str(tmp_path),
)
assert len(attempts) == 1
assert response.status_code == status_code
assert response.json() == body
assert list(tmp_path.iterdir()) == []
4 changes: 2 additions & 2 deletions src/unstructured_client/_version.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,10 @@
import importlib.metadata

__title__: str = "unstructured-client"
__version__: str = "0.46.3"
__version__: str = "0.46.4"
__openapi_doc_version__: str = "1.2.31"
__gen_version__: str = "2.680.0"
__user_agent__: str = "speakeasy-sdk/python 0.46.3 2.680.0 1.2.31 unstructured-client"
__user_agent__: str = "speakeasy-sdk/python 0.46.4 2.680.0 1.2.31 unstructured-client"

try:
if __package__ is not None:
Expand Down
51 changes: 25 additions & 26 deletions src/unstructured_client/utils/retries.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,27 @@ def __init__(self, inner: Exception):
self.inner = inner


def _declares_non_retryable(response: httpx.Response) -> bool:
# Inspect buffered error bodies without consuming unread response streams.
try:
body = response.json()
except (ValueError, RecursionError, httpx.ResponseNotRead):
Comment thread
CyMule marked this conversation as resolved.
return False
return isinstance(body, dict) and body.get("retryable") is False


def _is_retry_status(response: httpx.Response, status_codes: List[str]) -> bool:
for code in status_codes:
if "X" in code.upper():
code_range = int(code[0])
status_major = response.status_code / 100
if code_range <= status_major < code_range + 1:
return True
elif response.status_code == int(code):
return True
return False


def retry(func, retries: Retries):
if retries.config.strategy == "backoff":

Expand All @@ -95,19 +116,8 @@ def do_request() -> httpx.Response:
try:
res = func()

for code in retries.status_codes:
if "X" in code.upper():
code_range = int(code[0])

status_major = res.status_code / 100

if code_range <= status_major < code_range + 1:
raise TemporaryError(res)
else:
parsed_code = int(code)

if res.status_code == parsed_code:
raise TemporaryError(res)
if _is_retry_status(res, retries.status_codes) and not _declares_non_retryable(res):
raise TemporaryError(res)
except httpx.TransportError as exception:
if retries.config.retry_connection_errors:
raise
Expand Down Expand Up @@ -141,19 +151,8 @@ async def do_request() -> httpx.Response:
try:
res = await func()

for code in retries.status_codes:
if "X" in code.upper():
code_range = int(code[0])

status_major = res.status_code / 100

if code_range <= status_major < code_range + 1:
raise TemporaryError(res)
else:
parsed_code = int(code)

if res.status_code == parsed_code:
raise TemporaryError(res)
if _is_retry_status(res, retries.status_codes) and not _declares_non_retryable(res):
raise TemporaryError(res)
except httpx.TransportError as exception:
if retries.config.retry_connection_errors:
raise
Expand Down
Loading