fix(subscriptions): refuse malformed list/events 200s and replaying cursors (#142)

CHANGELOG.md

@@ -71,6 +71,23 @@ All notable changes to the OilPriceAPI Python SDK will be documented in this fil
 
 ### Fixed
 
+- **`subscriptions.list()` and `subscriptions.events()` no longer report a
+  malformed success as "nothing there" (#142), sync and async.** A 200 without
+  a `data.subscriptions` list returned `[]`, and one without `data.events` /
+  `data.cursor` returned an empty page with `cursor=None`. Fed back as
+  `events(since=page.cursor)`, that `None` dropped `since`, and the API reads a
+  missing `since` as `0`, so the poller replayed the account's whole event
+  history. Both now raise `OilPriceAPIError(code="MALFORMED_RESPONSE")` with the
+  raw body when the collection is missing or mistyped, a record is invalid,
+  `cursor` is not a non-negative integer, `has_more` is not a boolean, or the
+  cursor is behind `since` or behind an event in the page. A genuinely empty
+  list or page is still an empty success, and `page.cursor` is now always an
+  `int`.
+- **`subscriptions.events(since=...)` refuses a cursor the API would read as
+  `0`.** The API parses `since` with `to_i`, so `"abc"`, `""` and `-1` replay
+  every event and `1.5` becomes `1` (verified against production on
+  2026-09-13). Anything but a non-negative `int` or `None` now raises
+  `ValidationError(field="since", status_code=None)` before a request is sent.
 - **`subscriptions.create()` no longer turns a malformed success into a
   half-built record.** It fell back to treating the whole `data` object as the
   subscription when `data.subscription` was missing, and leaked a raw pydantic

oilpriceapi/_subscriptions_common.py

@@ -14,6 +14,7 @@
 
 if TYPE_CHECKING:
     from .models import Subscription
+    from .resources.subscriptions import SubscriptionEventsPage
 
 # Default attribution source stamped on subscriptions created via this SDK.
 DEFAULT_SOURCE = "sdk-python"
@@ -251,6 +252,152 @@ def unwrap_subscription(response: Any, *, subject: str) -> "Subscription":
         ) from error
 
 
+def _field_errors(error: Any) -> str:
+    return ", ".join(".".join(str(part) for part in item["loc"]) for item in error.errors())
+
+
+def unwrap_subscription_list(response: Any, *, subject: str) -> List["Subscription"]:
+    """Return the typed ``data.subscriptions`` list from a success body.
+
+    ``GET /v1/subscriptions`` always answers
+    ``{"status": "success", "data": {"subscriptions": [...]}}``. An empty list is
+    returned only when the API sent an empty list.
+
+    Raises:
+        OilPriceAPIError: ``code="MALFORMED_RESPONSE"`` when ``data.subscriptions``
+            is missing or not a list, or a record in it is not a valid
+            subscription. A malformed success is never reported as "no
+            subscriptions".
+    """
+    from pydantic import ValidationError as PydanticValidationError
+
+    from ._fuel_surcharge_common import _malformed
+    from .models import Subscription
+
+    data = response.get("data") if isinstance(response, dict) else None
+    records = data.get("subscriptions") if isinstance(data, dict) else None
+    if not isinstance(records, list):
+        raise _malformed(subject, "expected data.subscriptions to be a list", response)
+
+    subscriptions: List[Subscription] = []
+    for index, record in enumerate(records):
+        if not isinstance(record, dict):
+            raise _malformed(
+                subject, f"expected data.subscriptions[{index}] to be an object", response
+            )
+        try:
+            subscriptions.append(Subscription(**record))
+        except PydanticValidationError as error:
+            raise _malformed(
+                subject,
+                f"data.subscriptions[{index}] has {error.error_count()} invalid or missing "
+                f"field(s): {_field_errors(error)}",
+                response,
+            ) from error
+    return subscriptions
+
+
+def validate_since(since: Any) -> Optional[int]:
+    """Return ``since`` if the API will read it as the cursor it is.
+
+    ``GET /v1/subscriptions/events`` reads ``params[:since].to_i``: a blank,
+    non-numeric or negative value becomes ``0`` and replays every event the
+    account has, and ``1.5`` becomes ``1``. ``None`` (omitted) is the first
+    poll; anything else must be a previous page's integer ``cursor``, or ``0``.
+
+    Raises:
+        ValidationError: ``field="since"``, ``status_code=None``. Nothing is sent.
+    """
+    if since is None:
+        return None
+    if isinstance(since, bool) or not isinstance(since, int) or since < 0:
+        raise _refuse(
+            f"Invalid events cursor since={since!r}: pass page.cursor from the previous "
+            f"events() call, 0 to start from the first event, or omit it on the first "
+            f"poll. The API reads any other value as 0 and replays every event.",
+            "since",
+            since,
+        )
+    return since
+
+
+def unwrap_events_page(
+    response: Any, *, since: Optional[int], subject: str
+) -> "SubscriptionEventsPage":
+    """Return the typed events page from a ``GET /v1/subscriptions/events`` body.
+
+    The API always sends ``data.cursor`` (an integer: the last event's ``seq``,
+    or ``since`` when there are none), ``data.has_more`` and ``data.events``.
+    The cursor is what the caller sends as ``since`` next, so a missing or
+    wrong-typed cursor is refused rather than defaulted: ``cursor=None`` would
+    make the next poll omit ``since`` and restart from the first event.
+
+    Raises:
+        OilPriceAPIError: ``code="MALFORMED_RESPONSE"`` when ``data.events`` is
+            not a list of events, ``data.cursor`` is not a non-negative integer,
+            ``data.has_more`` is not a boolean, or the cursor is behind ``since``
+            or behind an event in the page (following it would replay events).
+    """
+    from pydantic import ValidationError as PydanticValidationError
+
+    from ._fuel_surcharge_common import _malformed
+    from .models import SubscriptionEvent
+    from .resources.subscriptions import SubscriptionEventsPage
+
+    data = response.get("data") if isinstance(response, dict) else None
+    if not isinstance(data, dict):
+        raise _malformed(subject, "expected a 'data' object", response)
+
+    records = data.get("events")
+    if not isinstance(records, list):
+        raise _malformed(subject, "expected data.events to be a list", response)
+
+    cursor = data.get("cursor")
+    if isinstance(cursor, bool) or not isinstance(cursor, int) or cursor < 0:
+        raise _malformed(
+            subject,
+            f"expected data.cursor to be a non-negative integer, got {cursor!r}",
+            response,
+        )
+
+    has_more = data.get("has_more")
+    if not isinstance(has_more, bool):
+        raise _malformed(
+            subject, f"expected data.has_more to be true or false, got {has_more!r}", response
+        )
+
+    if since is not None and cursor < since:
+        raise _malformed(
+            subject,
+            f"data.cursor {cursor} is behind since={since}; following it would replay events",
+            response,
+        )
+
+    events: List[SubscriptionEvent] = []
+    for index, record in enumerate(records):
+        if not isinstance(record, dict):
+            raise _malformed(subject, f"expected data.events[{index}] to be an object", response)
+        try:
+            event = SubscriptionEvent(**record)
+        except PydanticValidationError as error:
+            raise _malformed(
+                subject,
+                f"data.events[{index}] has {error.error_count()} invalid field(s): "
+                f"{_field_errors(error)}",
+                response,
+            ) from error
+        if event.seq is not None and event.seq > cursor:
+            raise _malformed(
+                subject,
+                f"data.cursor {cursor} is behind data.events[{index}].seq {event.seq}; "
+                f"following it would replay events",
+                response,
+            )
+        events.append(event)
+
+    return SubscriptionEventsPage(events=events, cursor=cursor, has_more=has_more)
+
+
 def unwrap_data(response: Any) -> Dict[str, Any]:
     """Return the ``data`` object from a ``{status, data}`` envelope."""
     if isinstance(response, dict) and "data" in response:

oilpriceapi/async_resources.py

@@ -8,8 +8,10 @@
     build_attribution_headers,
     build_create_body,
     build_update_body,
-    unwrap_data,
+    unwrap_events_page,
     unwrap_subscription,
+    unwrap_subscription_list,
+    validate_since,
     validate_subscription_id,
 )
 from .exceptions import ValidationError
@@ -42,7 +44,6 @@
     ParcelFuelSurchargeCarrier,
     PriceAlert,
     Subscription,
-    SubscriptionEvent,
 )
 from .resource_validators import (
     VALID_OPERATORS,
@@ -1600,11 +1601,13 @@ def __init__(self, client: Any) -> None:
         self.client = client
 
     async def list(self) -> List[Subscription]:
-        """List all subscriptions for the authenticated user."""
+        """List all subscriptions. See ``SubscriptionsResource.list``.
+
+        Raises ``OilPriceAPIError(code="MALFORMED_RESPONSE")`` when a success
+        body has no ``data.subscriptions`` list; empty only when the API sent [].
+        """
         response = await self.client.request(method="GET", path="/v1/subscriptions")
-        data = unwrap_data(response)
-        subs = data.get("subscriptions", [])
-        return [Subscription(**s) for s in subs]
+        return unwrap_subscription_list(response, subject="subscriptions.list")
 
     async def create(
         self,
@@ -1706,8 +1709,12 @@ async def events(
     ) -> SubscriptionEventsPage:
         """Poll for subscription events newer than a cursor.
 
-        Returns a SubscriptionEventsPage with events, cursor, and has_more.
+        See ``SubscriptionsResource.events``. ``since`` must be a non-negative
+        int (``ValidationError``, nothing sent, otherwise), and a success body
+        without an integer cursor raises ``MALFORMED_RESPONSE`` rather than
+        returning ``cursor=None``, which would restart polling from event 1.
         """
+        since = validate_since(since)
         params: Dict[str, Any] = {}
         if since is not None:
             params["since"] = since
@@ -1721,11 +1728,7 @@ async def events(
             path="/v1/subscriptions/events",
             params=params,
         )
-        data = unwrap_data(response)
-        events = [SubscriptionEvent(**e) for e in data.get("events", [])]
-        cursor = data.get("cursor")
-        has_more = bool(data.get("has_more", False))
-        return SubscriptionEventsPage(events=events, cursor=cursor, has_more=has_more)
+        return unwrap_events_page(response, since=since, subject="subscriptions.events")
 
 
 class AsyncFuelSurchargeResource:

oilpriceapi/resources/subscriptions.py

@@ -5,14 +5,16 @@
 periodically evaluate commodity codes and emit events an agent can poll for.
 """
 
-from typing import Any, Dict, List, Optional, Union, cast
+from typing import Any, Dict, List, Optional, Union
 
 from .._subscriptions_common import (
     build_attribution_headers,
     build_create_body,
     build_update_body,
-    unwrap_data,
+    unwrap_events_page,
     unwrap_subscription,
+    unwrap_subscription_list,
+    validate_since,
     validate_subscription_id,
 )
 from ..models import Subscription, SubscriptionEvent
@@ -57,16 +59,19 @@ def list(self) -> List[Subscription]:
         """List all subscriptions for the authenticated user.
 
         Returns:
-            List of Subscription models.
+            List of Subscription models. Empty only when the API sent an empty
+            list.
+
+        Raises:
+            OilPriceAPIError: ``code="MALFORMED_RESPONSE"`` when a success body
+                has no ``data.subscriptions`` list or a record in it is invalid.
 
         Example:
             >>> for sub in client.subscriptions.list():
             ...     print(sub.name, sub.codes)
         """
         response = self.client.request(method="GET", path="/v1/subscriptions")
-        data = unwrap_data(response)
-        subs = data.get("subscriptions", [])
-        return [Subscription(**s) for s in subs]
+        return unwrap_subscription_list(response, subject="subscriptions.list")
 
     def create(
         self,
@@ -254,19 +259,32 @@ def events(
         """Poll for subscription events newer than a cursor.
 
         Args:
-            since: Sequence cursor; only events with seq > since are returned.
+            since: ``page.cursor`` from the previous call; only events with
+                ``seq > since`` are returned. Omit it (or pass ``0``) only to
+                start from the first event. Must be a non-negative ``int``: the
+                API reads any other value as ``0`` and replays every event.
             limit: Max events to return (server clamps to its own max).
             watch_id: Restrict to a single subscription.
 
         Returns:
-            A SubscriptionEventsPage with events, cursor, and has_more.
+            A SubscriptionEventsPage with events, cursor, and has_more. The
+            cursor is always an ``int``, so ``events(since=page.cursor)`` never
+            restarts from the beginning.
+
+        Raises:
+            ValidationError: ``field="since"``, ``status_code=None``, if ``since``
+                is not a non-negative int. Nothing is sent.
+            OilPriceAPIError: ``code="MALFORMED_RESPONSE"`` when a success body
+                lacks the ``events`` list, an integer ``cursor`` or a boolean
+                ``has_more``, or its cursor would move polling backwards.
 
         Example:
             >>> page = client.subscriptions.events(since=0)
             >>> for event in page:
-            ...     print(event.type, event.code)
+            ...     print(event.seq, event.watch_id)
             >>> next_page = client.subscriptions.events(since=page.cursor)
         """
+        since = validate_since(since)
         params: Dict[str, Any] = {}
         if since is not None:
             params["since"] = since
@@ -280,8 +298,4 @@ def events(
             path="/v1/subscriptions/events",
             params=params,
         )
-        data = unwrap_data(response)
-        events = [SubscriptionEvent(**e) for e in data.get("events", [])]
-        cursor = cast(Optional[int], data.get("cursor"))
-        has_more = bool(data.get("has_more", False))
-        return SubscriptionEventsPage(events=events, cursor=cursor, has_more=has_more)
+        return unwrap_events_page(response, since=since, subject="subscriptions.events")