polish: cursor guard O(1); batch bytes path; bench scripts; close terminality; env isolation; from_env test

kalshi/_base_client.py

@@ -178,6 +178,8 @@ def request(
         *,
         params: dict[str, Any] | None = None,
         json: dict[str, Any] | None = None,
+        content: bytes | None = None,
+        headers: dict[str, str] | None = None,
         extra_headers: dict[str, str] | None = None,
     ) -> httpx.Response:
         """Make an authenticated HTTP request with retry logic.
@@ -187,6 +189,8 @@ def request(
         client-default one), but the ``KALSHI-ACCESS-*`` auth headers always
         win — callers cannot forge them via ``extra_headers``.
         """
+        if json is not None and content is not None:
+            raise TypeError("request() accepts `json=` or `content=`, not both.")
         # P1.6: canonicalize trailing slash BEFORE both signing and httpx call
         # so the wire path and the signed payload stay byte-identical. Root "/"
         # is preserved.
@@ -200,11 +204,18 @@ def request(
         total_timeout = self._config.total_timeout
         config_extra = self._config.extra_headers or {}
         per_call_extra = extra_headers or {}
+        body_headers = headers or {}
 
         for attempt in range(self._config.max_retries + 1):
             auth_headers = self._auth.sign_request(method.upper(), sign_path) if self._auth else {}
-            # Order matters: config defaults < per-request overrides < signed auth.
-            merged_headers = {**config_extra, **per_call_extra, **auth_headers}
+            # Order matters: config defaults < per-request overrides
+            #               < body-helper headers < signed auth.
+            merged_headers = {
+                **config_extra,
+                **per_call_extra,
+                **body_headers,
+                **auth_headers,
+            }
 
             logger.debug(
                 "Request: %s %s (attempt %d/%d)",
@@ -220,6 +231,7 @@ def request(
                     url=path,
                     params=params,
                     json=json,
+                    content=content,
                     headers=merged_headers,
                 )
             except httpx.PoolTimeout as e:
@@ -349,6 +361,8 @@ async def request(
         *,
         params: dict[str, Any] | None = None,
         json: dict[str, Any] | None = None,
+        content: bytes | None = None,
+        headers: dict[str, str] | None = None,
         extra_headers: dict[str, str] | None = None,
     ) -> httpx.Response:
         """Make an authenticated async HTTP request with retry logic.
@@ -357,6 +371,9 @@ async def request(
         """
         import asyncio
 
+        if json is not None and content is not None:
+            raise TypeError("request() accepts `json=` or `content=`, not both.")
+
         # P1.6: canonicalize trailing slash BEFORE both signing and httpx call.
         if path != "/" and path.endswith("/"):
             path = path.rstrip("/") or "/"
@@ -368,13 +385,19 @@ async def request(
         total_timeout = self._config.total_timeout
         config_extra = self._config.extra_headers or {}
         per_call_extra = extra_headers or {}
+        body_headers = headers or {}
 
         for attempt in range(self._config.max_retries + 1):
             if self._auth:
                 auth_headers = await self._auth.sign_request_async(method.upper(), sign_path)
             else:
                 auth_headers = {}
-            merged_headers = {**config_extra, **per_call_extra, **auth_headers}
+            merged_headers = {
+                **config_extra,
+                **per_call_extra,
+                **body_headers,
+                **auth_headers,
+            }
 
             logger.debug(
                 "Async request: %s %s (attempt %d/%d)",
@@ -390,6 +413,7 @@ async def request(
                     url=path,
                     params=params,
                     json=json,
+                    content=content,
                     headers=merged_headers,
                 )
             except httpx.PoolTimeout as e:

kalshi/auth.py

@@ -71,6 +71,9 @@ def __init__(self, key_id: str, private_key: rsa.RSAPrivateKey) -> None:
         # await-latency rather than inline CPU. (#178)
         self._sign_executor: ThreadPoolExecutor | None = None
         self._sign_executor_lock = threading.Lock()
+        # Set in close(); checked by _get_sign_executor so async signs after
+        # close() raise instead of silently respawning the executor.
+        self._closed: bool = False
 
     @classmethod
     def from_key_path(
@@ -278,7 +281,17 @@ def _get_sign_executor(self) -> ThreadPoolExecutor:
 
         Thread-safe lazy init: the lock guards the assignment, and the
         double-checked pattern keeps the fast path lock-free after first use.
+
+        Raises ``RuntimeError`` after :meth:`close` so a closed auth does
+        not silently respawn the executor (which would defeat the point of
+        ``close()`` — the old executor's threads stay shut down but a new
+        ThreadPoolExecutor would be allocated on the next async sign).
         """
+        if self._closed:
+            raise RuntimeError(
+                "KalshiAuth has been closed; create a new instance to "
+                "sign further requests."
+            )
         if self._sign_executor is None:
             with self._sign_executor_lock:
                 if self._sign_executor is None:
@@ -312,14 +325,20 @@ async def sign_request_async(
         )
 
     def close(self) -> None:
-        """Shut down the sign-offload executor if one was created.
+        """Shut down the sign-offload executor and mark this auth closed.
 
         Idempotent. Safe to call without an executor having been initialised.
         Long-lived clients ordinarily rely on interpreter-shutdown cleanup
         (atexit-joined executor threads); this hook is for tests and other
         callers that want deterministic teardown.
+
+        Terminal: a subsequent :meth:`sign_request_async` raises
+        ``RuntimeError`` rather than silently respawning a fresh executor.
+        The synchronous :meth:`sign_request` is unaffected — it never used
+        the executor — but callers should treat the instance as retired.
         """
         with self._sign_executor_lock:
+            self._closed = True
             if self._sign_executor is not None:
                 self._sign_executor.shutdown(wait=False)
                 self._sign_executor = None

kalshi/resources/_base.py

@@ -150,6 +150,26 @@ def _post(
         result: dict[str, Any] = response.json()
         return result
 
+    def _post_json(
+        self,
+        path: str,
+        *,
+        content: bytes,
+        params: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """POST a pre-serialized JSON body (bytes).
+
+        Skips the dict-walk pass httpx's ``json=`` does. Use this when
+        a Pydantic model already produced the wire bytes via
+        ``model_dump_json(...).encode()``.
+        """
+        response = self._transport.request(
+            "POST", path, params=params, content=content,
+            headers={"Content-Type": "application/json"},
+        )
+        result: dict[str, Any] = response.json()
+        return result
+
     def _put(
         self,
         path: str,
@@ -181,6 +201,23 @@ def _delete_with_body(
         result: dict[str, Any] = response.json()
         return result
 
+    def _delete_with_body_json(
+        self,
+        path: str,
+        *,
+        content: bytes,
+        params: dict[str, Any] | None = None,
+    ) -> dict[str, Any] | None:
+        """DELETE with a pre-serialized JSON body (bytes). See :meth:`_post_json`."""
+        response = self._transport.request(
+            "DELETE", path, params=params, content=content,
+            headers={"Content-Type": "application/json"},
+        )
+        if response.status_code == 204:
+            return None
+        result: dict[str, Any] = response.json()
+        return result
+
     def _list(
         self,
         path: str,
@@ -225,11 +262,13 @@ def _list_all(
         cursor-repeat guard below provides the safety net against
         infinite loops.
 
-        Raises ``KalshiError`` if a cursor value repeats, which indicates
-        a server-side pagination bug.
+        Raises ``KalshiError`` if the previous page's cursor repeats —
+        the realistic server-pagination-bug shape (last page replayed
+        forever). Only the most recent cursor is retained, so memory is
+        O(1) regardless of page count.
         """
         current_params = dict(params) if params else {}
-        seen_cursors: set[str] = set()
+        last_cursor: str | None = None
         pages_fetched = 0
         while max_pages is None or pages_fetched < max_pages:
             page = self._list(
@@ -240,13 +279,13 @@ def _list_all(
             pages_fetched += 1
             if not page.cursor:
                 break
-            if page.cursor in seen_cursors:
+            if page.cursor == last_cursor:
                 raise KalshiError(
                     f"Cursor loop detected on {path}: server returned "
                     f"cursor {page.cursor!r} which was already used. "
                     f"This indicates a server-side pagination bug."
                 )
-            seen_cursors.add(page.cursor)
+            last_cursor = page.cursor
             current_params["cursor"] = page.cursor
 
 
@@ -282,6 +321,21 @@ async def _post(
         result: dict[str, Any] = response.json()
         return result
 
+    async def _post_json(
+        self,
+        path: str,
+        *,
+        content: bytes,
+        params: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """Async :meth:`SyncResource._post_json`."""
+        response = await self._transport.request(
+            "POST", path, params=params, content=content,
+            headers={"Content-Type": "application/json"},
+        )
+        result: dict[str, Any] = response.json()
+        return result
+
     async def _put(
         self,
         path: str,
@@ -313,6 +367,23 @@ async def _delete_with_body(
         result: dict[str, Any] = response.json()
         return result
 
+    async def _delete_with_body_json(
+        self,
+        path: str,
+        *,
+        content: bytes,
+        params: dict[str, Any] | None = None,
+    ) -> dict[str, Any] | None:
+        """Async :meth:`SyncResource._delete_with_body_json`."""
+        response = await self._transport.request(
+            "DELETE", path, params=params, content=content,
+            headers={"Content-Type": "application/json"},
+        )
+        if response.status_code == 204:
+            return None
+        result: dict[str, Any] = response.json()
+        return result
+
     async def _list(
         self,
         path: str,
@@ -344,7 +415,7 @@ async def _list_all(
         Raises ``KalshiError`` on repeated cursor; see sync docstring.
         """
         current_params = dict(params) if params else {}
-        seen_cursors: set[str] = set()
+        last_cursor: str | None = None
         pages_fetched = 0
         while max_pages is None or pages_fetched < max_pages:
             page = await self._list(
@@ -356,11 +427,11 @@ async def _list_all(
             pages_fetched += 1
             if not page.cursor:
                 break
-            if page.cursor in seen_cursors:
+            if page.cursor == last_cursor:
                 raise KalshiError(
                     f"Cursor loop detected on {path}: server returned "
                     f"cursor {page.cursor!r} which was already used. "
                     f"This indicates a server-side pagination bug."
                 )
-            seen_cursors.add(page.cursor)
+            last_cursor = page.cursor
             current_params["cursor"] = page.cursor

kalshi/resources/orders.py

@@ -124,21 +124,29 @@ def _build_create_order_body(
 def _build_batch_create_body(
     request: BatchCreateOrdersRequest | None,
     orders: Sequence[CreateOrderRequest] | None,
-) -> dict[str, Any]:
+) -> bytes:
+    """Serialize the batch-create body directly to JSON bytes.
+
+    Skips the intermediate ``model_dump(mode="json")`` dict-walk: with up
+    to 100 orders x ~10 Decimal fields, the dict version pays the
+    serializer cost twice (once here, once in httpx's json encoder).
+    """
     _check_request_exclusive(request, orders=orders)
     if request is None:
         if orders is None:
             raise TypeError(
                 "batch_create() requires `orders` (or pass `request=...`)"
             )
         request = BatchCreateOrdersRequest(orders=list(orders))
-    return request.model_dump(exclude_none=True, by_alias=True, mode="json")
+    return request.model_dump_json(exclude_none=True, by_alias=True).encode()
 
 
 def _build_batch_cancel_body(
     request: BatchCancelOrdersRequest | None,
     orders: Sequence[BatchCancelOrdersRequestOrder | str] | None,
-) -> dict[str, Any]:
+) -> bytes:
+    """Serialize the batch-cancel body directly to JSON bytes. See
+    :func:`_build_batch_create_body` for the perf rationale."""
     _check_request_exclusive(request, orders=orders)
     if request is None:
         if orders is None:
@@ -152,7 +160,7 @@ def _build_batch_cancel_body(
             for o in orders
         ]
         request = BatchCancelOrdersRequest(orders=normalized)
-    return request.model_dump(exclude_none=True, by_alias=True, mode="json")
+    return request.model_dump_json(exclude_none=True, by_alias=True).encode()
 
 
 def _build_amend_body(
@@ -475,7 +483,7 @@ def batch_create(
         """
         self._require_auth()
         body = _build_batch_create_body(request, orders)
-        data = self._post("/portfolio/orders/batched", json=body)
+        data = self._post_json("/portfolio/orders/batched", content=body)
         return BatchCreateOrdersResponse.model_validate(data)
 
     @overload
@@ -513,7 +521,7 @@ def batch_cancel(
         """
         self._require_auth()
         body = _build_batch_cancel_body(request, orders)
-        data = self._delete_with_body("/portfolio/orders/batched", json=body)
+        data = self._delete_with_body_json("/portfolio/orders/batched", content=body)
         if data is None:
             raise KalshiError(
                 "Expected BatchCancelOrdersResponse body, got 204 No Content."
@@ -929,7 +937,7 @@ async def batch_create(
         """Place a batch of orders. See :meth:`OrdersResource.batch_create`."""
         self._require_auth()
         body = _build_batch_create_body(request, orders)
-        data = await self._post("/portfolio/orders/batched", json=body)
+        data = await self._post_json("/portfolio/orders/batched", content=body)
         return BatchCreateOrdersResponse.model_validate(data)
 
     @overload
@@ -952,7 +960,7 @@ async def batch_cancel(
         """Batch-cancel orders. See :meth:`OrdersResource.batch_cancel`."""
         self._require_auth()
         body = _build_batch_cancel_body(request, orders)
-        data = await self._delete_with_body("/portfolio/orders/batched", json=body)
+        data = await self._delete_with_body_json("/portfolio/orders/batched", content=body)
         if data is None:
             raise KalshiError(
                 "Expected BatchCancelOrdersResponse body, got 204 No Content."