Poll cached full scan results until ready (#115)

* Poll cached full scan results

Wait for cached full scan artifacts and bound retries so SDK callers receive completed results or a clear timeout.

Assisted-by: Codex: GPT-6

* Fix cached scan CI coverage

Exercise the cached stream through the real SDK transport and bump the package version to an unpublished release.

Assisted-by: Codex: GPT-6

* Tighten full scan polling and document it

Stop polling once the next attempt could not start before the deadline instead of sleeping out the budget, give each request the configured timeout rather than a fixed 30s cap, and document the polling parameters in the README.

---------

Co-authored-by: lelia <2418071+lelia@users.noreply.github.com>

Author: jdalton

CHANGELOG.md

@@ -6,6 +6,29 @@
 
 - The SDK rejects the retired API credential before storing or encoding it.
 
+### Fixed: full-scan reads wait for analysis to finish
+
+- `fullscans.stream()` now requests cached results (`?cached=true`) and polls
+  while the API answers `202 Processing`, instead of treating the first `202`
+  as a failed result. Polling backs off from 5s to 10s between attempts and
+  stops when the budget runs out, raising `APITimeout`.
+- The polling budget defaults to the SDK's configured `timeout`. Callers can
+  tune it with the new keyword-only arguments `poll_timeout_seconds`,
+  `poll_initial_interval_seconds` and `poll_max_interval_seconds`. Each
+  request is bounded by the configured `timeout` or the remaining budget,
+  whichever is shorter, and polling stops early once the next attempt could
+  not start before the budget runs out.
+- Transient failures (HTTP 408/502/503/504, timeouts, dropped connections) are
+  retried within the budget. Authorization, not-found and quota errors still
+  raise immediately.
+- Invalid polling settings (zero, negative, non-finite, booleans, or an
+  initial interval above the maximum) raise `ValueError` before any request is
+  sent.
+- The organization slug and full scan ID are now URL-encoded in the request
+  path.
+- `API.do_request()` accepts an optional keyword-only `request_timeout` that
+  overrides the configured timeout for a single call.
+
 ## 3.6.1
 
 ### Security: patched releases of certifi, uv and virtualenv

README.rst

@@ -291,7 +291,11 @@ Stream a diff between two full scans. Returns a scan diff.
 
 fullscans.stream(org_slug, full_scan_id, use_types=False)
 """""""""""""""""""""""""""""""""""""""""""""""""""""""""
-Stream all SBOM artifacts for a full scan.
+Stream all SBOM artifacts for a full scan. If the scan is still being analyzed, the SDK polls
+cached results until they are ready, backing off from 5 to 10 seconds between attempts, and
+raises ``APITimeout`` if they are not ready within the polling budget. Transient failures
+(HTTP 408/502/503/504, timeouts, dropped connections) are retried within the budget; other
+errors raise immediately.
 
 **Usage:**
 
@@ -304,11 +308,17 @@ Stream all SBOM artifacts for a full scan.
     # With typed response
     print(socket.fullscans.stream("org_slug", "full_scan_id", use_types=True))
 
+    # Wait up to 10 minutes for analysis to finish
+    print(socket.fullscans.stream("org_slug", "full_scan_id", poll_timeout_seconds=600))
+
 **PARAMETERS:**
 
 - **org_slug (str)** - The organization name
 - **full_scan_id (str)** - The ID of the full scan
 - **use_types (bool)** - Whether to return typed response objects (default: False)
+- **poll_timeout_seconds (float)** - Keyword-only. Total time to wait for results (default: the SDK's configured ``timeout``)
+- **poll_initial_interval_seconds (float)** - Keyword-only. Wait before the first retry (default: 5.0)
+- **poll_max_interval_seconds (float)** - Keyword-only. Longest wait between retries (default: 10.0)
 
 fullscans.metadata(org_slug, full_scan_id, use_types=False)
 """""""""""""""""""""""""""""""""""""""""""""""""""""""""""

socketdev/core/api.py

@@ -1,4 +1,5 @@
 import base64
+from typing import Optional
 from socketdev.log import log
 from socketdev.core.retired_token import reject_retired_basic_auth_token
 
@@ -49,6 +50,8 @@ def do_request(
         payload: [dict, str] = None,
         files: list = None,
         method: str = "GET",
+        *,
+        request_timeout: Optional[float] = None,
     ) -> Response:
         if self.encoded_key is None or self.encoded_key == "":
             raise APIKeyMissing
@@ -70,7 +73,8 @@ def format_headers(headers_dict):
 
             response = requests.request(
                 method.upper(), url, headers=headers, data=payload, files=files, 
-                timeout=self.request_timeout, verify=not self.allow_unverified
+                timeout=self.request_timeout if request_timeout is None else request_timeout,
+                verify=not self.allow_unverified
             )
             request_duration = time.time() - start_time
 

socketdev/fullscans/__init__.py

@@ -4,9 +4,12 @@
 from typing import Any, Dict, List, Optional, Union
 from dataclasses import dataclass, asdict, field
 import urllib.parse
+import math
+import time
 from ..core.dedupe import Dedupe
 from ..core.enums import unknown_enum_value
 from ..utils import IntegrationType, Utils
+from ..exceptions import APIFailure, APITimeout
 
 log = logging.getLogger("socketdev")
 
@@ -960,15 +963,65 @@ def stream_diff(
             )
         return {}
 
-    def stream(self, org_slug: str, full_scan_id: str, use_types: bool = False) -> Union[dict, FullScanStreamResponse]:
-        path = "orgs/" + org_slug + "/full-scans/" + full_scan_id
-        response = self.api.do_request(path=path, method="GET")
+    def stream(
+        self,
+        org_slug: str,
+        full_scan_id: str,
+        use_types: bool = False,
+        *,
+        poll_timeout_seconds: Optional[float] = None,
+        poll_initial_interval_seconds: float = 5.0,
+        poll_max_interval_seconds: float = 10.0,
+    ) -> Union[dict, FullScanStreamResponse]:
+        request_timeout = self.api.request_timeout
+        timeout = request_timeout if poll_timeout_seconds is None else poll_timeout_seconds
+        values = (timeout, poll_initial_interval_seconds, poll_max_interval_seconds, request_timeout)
+        if any(
+            isinstance(value, bool)
+            or not isinstance(value, (int, float))
+            or not math.isfinite(value)
+            or value <= 0
+            for value in values
+        ):
+            raise ValueError("Full scan polling timeout and intervals must be positive finite numbers")
+        if poll_initial_interval_seconds > poll_max_interval_seconds:
+            raise ValueError("Initial polling interval must not exceed maximum polling interval")
+
+        path = (
+            "orgs/" + urllib.parse.quote(org_slug, safe="") + "/full-scans/"
+            + urllib.parse.quote(full_scan_id, safe="") + "?cached=true"
+        )
+        deadline = time.monotonic() + timeout
+        interval = poll_initial_interval_seconds
+        while True:
+            retry_error = None
+            try:
+                response = self.api.do_request(
+                    path=path,
+                    method="GET",
+                    request_timeout=min(float(request_timeout), max(deadline - time.monotonic(), 0.001)),
+                )
+            except APIFailure as error:
+                if not error.is_transient_error():
+                    raise
+                retry_error = error
+            else:
+                if response.status_code != 202:
+                    break
+                response.close()
+            # Give up now if the next attempt could not start before the deadline,
+            # rather than sleeping out the rest of the budget and never polling again.
+            if deadline - time.monotonic() <= interval:
+                raise APITimeout("Timed out waiting for full scan results") from retry_error
+            time.sleep(interval)
+            interval = min(poll_max_interval_seconds, interval * 1.5)
 
         if response.status_code == 200:
             try:
                 stream_str = []
                 artifacts = {}
                 result = response.text
+                response.close()
                 result = result.strip('"').strip()
                 for line in result.split("\n"):
                     if line != '"' and line != "" and line is not None:
@@ -993,6 +1046,7 @@ def stream(self, org_slug: str, full_scan_id: str, use_types: bool = False) -> U
                 return artifacts
 
             except Exception as e:
+                response.close()
                 error_message = f"Error parsing stream response: {str(e)}"
                 log.error(error_message)
                 if use_types:
@@ -1001,7 +1055,10 @@ def stream(self, org_slug: str, full_scan_id: str, use_types: bool = False) -> U
                     )
                 return {}
 
-        error_message = response.json().get("error", {}).get("message", "Unknown error")
+        try:
+            error_message = response.json().get("error", {}).get("message", "Unknown error")
+        finally:
+            response.close()
         log.error(f"Error streaming full scan: {response.status_code}, message: {error_message}")
         if use_types:
             return FullScanStreamResponse.from_dict(

tests/support/socket_api_fixture.py

@@ -0,0 +1,121 @@
+"""Reusable localhost HTTP fixture for real Socket SDK transport tests."""
+
+from __future__ import annotations
+
+import json
+import threading
+from dataclasses import dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Callable, Dict, List, Optional, Tuple
+from urllib.parse import parse_qs, urlsplit
+
+
+@dataclass
+class SocketRequest:
+    method: str
+    path: str
+    query: Dict[str, List[str]]
+    headers: Dict[str, str]
+    body: bytes
+
+
+@dataclass
+class SocketResponse:
+    status: int
+    payload: object
+    raw_body: Optional[bytes] = None
+
+
+Route = Callable[[SocketRequest], SocketResponse]
+
+
+class SocketApiFixture:
+    def __init__(self) -> None:
+        self.requests: List[SocketRequest] = []
+        self._routes: Dict[Tuple[str, str], Route] = {}
+        self._server: Optional[ThreadingHTTPServer] = None
+        self._thread: Optional[threading.Thread] = None
+
+    @property
+    def base_url(self) -> str:
+        if self._server is None:
+            raise RuntimeError("Socket API fixture is not running")
+        host, port = self._server.server_address
+        return f"http://{host}:{port}/v0"
+
+    def route(
+        self,
+        method: str,
+        path: str,
+        response: SocketResponse | Route,
+    ) -> None:
+        if callable(response):
+            handler = response
+        else:
+
+            def handler(_request: SocketRequest) -> SocketResponse:
+                return response
+
+        self._routes[(method.upper(), path)] = handler
+
+    def start(self) -> None:
+        fixture = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def do_GET(self) -> None:
+                self._dispatch()
+
+            def do_POST(self) -> None:
+                self._dispatch()
+
+            def _dispatch(self) -> None:
+                parsed = urlsplit(self.path)
+                length = int(self.headers.get("Content-Length", "0"))
+                request = SocketRequest(
+                    method=self.command,
+                    path=parsed.path.removeprefix("/v0/"),
+                    query=parse_qs(parsed.query, keep_blank_values=True),
+                    headers=dict(self.headers.items()),
+                    body=self.rfile.read(length),
+                )
+                fixture.requests.append(request)
+                route = fixture._routes.get((request.method, request.path))
+                if route is None and request.path == "organizations":
+                    response = SocketResponse(
+                        200,
+                        {"organizations": {"org-fixture": {"slug": "fixture-org"}}},
+                    )
+                elif route is None:
+                    response = SocketResponse(404, {"error": {"message": "Not found"}})
+                else:
+                    response = route(request)
+                body = response.raw_body
+                if body is None:
+                    body = json.dumps(response.payload).encode("utf-8")
+                self.send_response(response.status)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+
+            def log_message(self, _format: str, *args: object) -> None:
+                return
+
+        self._server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
+        self._thread.start()
+
+    def close(self) -> None:
+        if self._server is not None:
+            self._server.shutdown()
+            self._server.server_close()
+            self._server = None
+        if self._thread is not None:
+            self._thread.join(timeout=2)
+            self._thread = None
+
+
+def socket_api_fixture() -> SocketApiFixture:
+    fixture = SocketApiFixture()
+    fixture.start()
+    return fixture