Poll cached full scan results until ready

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