create_from_paths: Allow complete retry

- Introduced `send_request_once` and `run_with_retry` for consistent
  request execution and retry handling.
- Implemented retries for `create_from_paths` in both sync and async
  clients, covering 429, transport, and timeout errors.
- Added concurrency control using semaphores for async file info fetches
  after uploads.
- Updated tests to validate retry behavior, including various error
  scenarios and proper retry counts.

Assisted-by: Codex

Author: datalogics-kam

src/pdfrest/client.py

@@ -896,6 +896,22 @@ def _send_request(self, request: _RequestModel) -> Any:
             should_continue=stream_retry_checker,
         )
 
+    def send_request_once(self, request: _RequestModel) -> Any:
+        return self._perform_request(self._client, request)
+
+    def run_with_retry(
+        self,
+        func: Callable[[], ReturnType],
+        *,
+        operation: str,
+        should_continue: Callable[[PdfRestError], bool] | None = None,
+    ) -> ReturnType:
+        return self._execute_with_retry(
+            func,
+            operation=operation,
+            should_continue=should_continue,
+        )
+
     def _perform_request(
         self, http_client: httpx.Client, request: _RequestModel
     ) -> Any:
@@ -1143,6 +1159,22 @@ async def _send_request(self, request: _RequestModel) -> Any:
             should_continue=stream_retry_checker,
         )
 
+    async def send_request_once(self, request: _RequestModel) -> Any:
+        return await self._perform_request(self._client, request)
+
+    async def run_with_retry(
+        self,
+        func: Callable[[], Awaitable[ReturnType]],
+        *,
+        operation: str,
+        should_continue: Callable[[PdfRestError], bool] | None = None,
+    ) -> ReturnType:
+        return await self._execute_with_retry(
+            func,
+            operation=operation,
+            should_continue=should_continue,
+        )
+
     async def _perform_request(
         self, http_client: httpx.AsyncClient, request: _RequestModel
     ) -> Any:
@@ -1441,25 +1473,42 @@ def create_from_paths(
         closed once the request completes.
         """
         normalized_paths = _normalize_path_inputs(file_paths)
-
-        with ExitStack() as stack:
-            upload_specs: list[FileTypes] = []
-            for spec in normalized_paths:
-                path, content_type, headers = _parse_path_spec(spec)
-                file_obj = stack.enter_context(path.open("rb"))
-                filename = path.name
-                if headers:
-                    upload_specs.append((filename, file_obj, content_type, headers))
-                elif content_type is not None:
-                    upload_specs.append((filename, file_obj, content_type))
-                else:
-                    upload_specs.append((filename, file_obj))
-            return self.create(
-                upload_specs,
-                extra_query=extra_query,
-                extra_headers=extra_headers,
-                timeout=timeout,
-            )
+        path_specs = [_parse_path_spec(spec) for spec in normalized_paths]
+
+        def attempt() -> list[PdfRestFile]:
+            with ExitStack() as stack:
+                upload_specs: list[FileTypes] = []
+                for path, content_type, headers in path_specs:
+                    file_obj = stack.enter_context(path.open("rb"))
+                    filename = path.name
+                    if headers:
+                        upload_specs.append((filename, file_obj, content_type, headers))
+                    elif content_type is not None:
+                        upload_specs.append((filename, file_obj, content_type))
+                    else:
+                        upload_specs.append((filename, file_obj))
+
+                request = self._client.prepare_request(
+                    "POST",
+                    "/upload",
+                    files=upload_specs,
+                    extra_query=extra_query,
+                    extra_headers=extra_headers,
+                    timeout=timeout,
+                )
+                payload = self._client.send_request_once(request)
+                file_ids = _extract_uploaded_file_ids(payload)
+                return [
+                    self._client.fetch_file_info(
+                        file_id,
+                        extra_query=extra_query,
+                        extra_headers=extra_headers,
+                        timeout=timeout,
+                    )
+                    for file_id in file_ids
+                ]
+
+        return self._client.run_with_retry(attempt, operation="POST /upload (paths)")
 
     def create_from_urls(
         self,
@@ -1681,25 +1730,54 @@ async def create_from_paths(
         closed once the request completes.
         """
         normalized_paths = _normalize_path_inputs(file_paths)
-
-        with ExitStack() as stack:
-            upload_specs: list[FileTypes] = []
-            for spec in normalized_paths:
-                path, content_type, headers = _parse_path_spec(spec)
-                file_obj = stack.enter_context(path.open("rb"))
-                filename = path.name
-                if headers:
-                    upload_specs.append((filename, file_obj, content_type, headers))
-                elif content_type is not None:
-                    upload_specs.append((filename, file_obj, content_type))
-                else:
-                    upload_specs.append((filename, file_obj))
-            return await self.create(
-                upload_specs,
-                extra_query=extra_query,
-                extra_headers=extra_headers,
-                timeout=timeout,
-            )
+        path_specs = [_parse_path_spec(spec) for spec in normalized_paths]
+
+        async def attempt() -> list[PdfRestFile]:
+            with ExitStack() as stack:
+                upload_specs: list[FileTypes] = []
+                for path, content_type, headers in path_specs:
+                    file_obj = stack.enter_context(path.open("rb"))
+                    filename = path.name
+                    if headers:
+                        upload_specs.append((filename, file_obj, content_type, headers))
+                    elif content_type is not None:
+                        upload_specs.append((filename, file_obj, content_type))
+                    else:
+                        upload_specs.append((filename, file_obj))
+
+                request = self._client.prepare_request(
+                    "POST",
+                    "/upload",
+                    files=upload_specs,
+                    extra_query=extra_query,
+                    extra_headers=extra_headers,
+                    timeout=timeout,
+                )
+                payload = await self._client.send_request_once(request)
+                file_ids = _extract_uploaded_file_ids(payload)
+                results: list[PdfRestFile] = []
+                semaphore = asyncio.Semaphore(self._concurrency_limit)
+
+                async def throttled_fetch(file_id: str) -> PdfRestFile:
+                    async with semaphore:
+                        return await self._client.fetch_file_info(
+                            file_id,
+                            extra_query=extra_query,
+                            extra_headers=extra_headers,
+                            timeout=timeout,
+                        )
+
+                if file_ids:
+                    results = list(
+                        await asyncio.gather(
+                            *(throttled_fetch(fid) for fid in file_ids)
+                        )
+                    )
+                return results
+
+        return await self._client.run_with_retry(
+            attempt, operation="POST /upload (paths)"
+        )
 
     async def create_from_urls(
         self,