Handle ENLYZE timeseries pagination

src/production_analytics/enlyze/gateway.py

@@ -68,52 +68,68 @@ class EnlyzeApiGateway:
                 raise ValueError(f"{name} must be timezone-aware")
         if end < start:
             raise ValueError("end must not precede start")
-        response = self._client.post_json(
+        request_body = {
-            "/v2/timeseries",
+            "machine": machine_id,
-            {
+            "start": start.astimezone(UTC).isoformat(),
-                "machine": machine_id,
+            "end": end.astimezone(UTC).isoformat(),
-                "start": start.astimezone(UTC).isoformat(),
+            "variables": [
-                "end": end.astimezone(UTC).isoformat(),
+                {"uuid": rate_variable_id},
-                "variables": [
+                {"uuid": gate_variable_id},
-                    {"uuid": rate_variable_id},
+            ],
-                    {"uuid": gate_variable_id},
+        }
-                ],
+        samples = []
-            },
+        followed_cursors: set[str] = set()
-        )
+        page = 1
-
+        while True:
-        try:
+            response = self._client.post_json("/v2/timeseries", request_body)
-            data = response.body["data"]
+            try:
-            columns = data["columns"]
+                data = response.body["data"]
-            if not isinstance(columns, list):
+                columns = data["columns"]
-                raise ValueError("columns must be a list")
+                if not isinstance(columns, list):
-            for required in ("time", rate_variable_id, gate_variable_id):
+                    raise ValueError("columns must be a list")
-                if columns.count(required) != 1:
+                for required in ("time", rate_variable_id, gate_variable_id):
-                    raise ValueError(f"required column {required!r} must occur exactly once")
+                    if columns.count(required) != 1:
-            time_index = columns.index("time")
+                        raise ValueError(f"required column {required!r} must occur exactly once")
-            rate_index = columns.index(rate_variable_id)
+                time_index = columns.index("time")
-            gate_index = columns.index(gate_variable_id)
+                rate_index = columns.index(rate_variable_id)
-            if not isinstance(data["records"], list):
+                gate_index = columns.index(gate_variable_id)
-                raise ValueError("records must be a list")
+                if not isinstance(data["records"], list):
-            samples = []
+                    raise ValueError("records must be a list")
-            for index, record in enumerate(data["records"]):
+                for index, record in enumerate(data["records"]):
-                try:
+                    try:
-                    if not isinstance(record, list) or len(record) != len(columns):
+                        if not isinstance(record, list) or len(record) != len(columns):
-                        raise ValueError("record must match columns")
+                            raise ValueError("record must match columns")
-                    rate, gate = record[rate_index], record[gate_index]
+                        rate, gate = record[rate_index], record[gate_index]
-                    for value in (rate, gate):
+                        for value in (rate, gate):
-                        if isinstance(value, bool) or not isinstance(value, (int, float)):
+                            if isinstance(value, bool) or not isinstance(value, (int, float)):
-                            raise ValueError("rate and gate must be numeric")
+                                raise ValueError("rate and gate must be numeric")
-                        if not isfinite(value):
+                            if not isfinite(value):
-                            raise ValueError("rate and gate must be finite")
+                                raise ValueError("rate and gate must be finite")
-                    samples.append(MaterialSample(
+                        samples.append(MaterialSample(
-                        timestamp=_parse_timestamp(record[time_index]),
+                            timestamp=_parse_timestamp(record[time_index]),
-                        material_rate_kg_per_hour=float(rate), gate_value=float(gate),
+                            material_rate_kg_per_hour=float(rate), gate_value=float(gate),
-                    ))
+                        ))
-                except (TypeError, ValueError) as exc:
+                    except (TypeError, ValueError) as exc:
-                    raise ValueError(f"malformed record {index}: {exc}") from exc
+                        raise ValueError(f"malformed record {index}: {exc}") from exc
-            return samples
+                next_cursor = None
-        except (KeyError, TypeError, ValueError) as exc:
+                if "metadata" in response.body:
-            raise ValueError(f"Invalid timeseries response: {exc}") from exc
+                    metadata = response.body["metadata"]
+                    if not isinstance(metadata, dict) or "next_cursor" not in metadata:
+                        raise ValueError("metadata must be an object with a next_cursor field")
+                    next_cursor = metadata["next_cursor"]
+                    if next_cursor is not None and (
+                        not isinstance(next_cursor, str) or not next_cursor
+                    ):
+                        raise ValueError("next_cursor must be null or a non-empty string")
+                if next_cursor is None:
+                    return samples
+                if next_cursor in followed_cursors:
+                    raise ValueError("next_cursor has already been followed")
+                followed_cursors.add(next_cursor)
+                request_body = {**request_body, "cursor": next_cursor}
+                page += 1
+            except (KeyError, TypeError, ValueError) as exc:
+                raise ValueError(f"Invalid timeseries response on page {page}: {exc}") from exc
 
 
 def _parse_timestamp(value: str) -> datetime:

tests/test_enlyze_gateway.py

@@ -202,3 +202,91 @@ def test_malformed_run_fails_clearly(item) -> None:
     client.get.return_value.body = {"data": [item]}
     with pytest.raises(ValueError, match="Invalid production-run response"):
         EnlyzeApiGateway(client).get_open_production_run("m")
+
+
+def test_timeseries_with_null_cursor(timeseries) -> None:
+    gateway, client, args = timeseries
+    client.post_json.return_value.body["metadata"] = {"next_cursor": None}
+    client.post_json.return_value.body["data"]["records"] = [
+        ["2026-09-03T00:00:00Z", 10, 2],
+    ]
+    samples = gateway.get_material_samples(**args)
+    assert len(samples) == 1
+    assert samples[0].material_rate_kg_per_hour == 10
+    client.post_json.assert_called_once()
+
+
+def test_timeseries_pagination_resolves_columns_per_page(timeseries) -> None:
+    gateway, client, args = timeseries
+    client.post_json.side_effect = [
+        Mock(body={
+            "data": {"columns": ["time", "rate", "gate"],
+                     "records": [["2026-09-03T00:00:00Z", 10, 2]]},
+            "metadata": {"next_cursor": "continuation"},
+        }),
+        Mock(body={
+            "data": {"columns": ["gate", "time", "rate"],
+                     "records": [[3, "2026-09-03T01:00:00Z", 20]]},
+            "metadata": {"next_cursor": None},
+        }),
+    ]
+    samples = gateway.get_material_samples(**args)
+    assert [(s.timestamp, s.material_rate_kg_per_hour, s.gate_value) for s in samples] == [
+        (datetime(2026, 9, 3, tzinfo=UTC), 10, 2),
+        (datetime(2026, 9, 3, 1, tzinfo=UTC), 20, 3),
+    ]
+    assert client.post_json.call_count == 2
+    first, second = client.post_json.call_args_list
+    assert first.args == ("/v2/timeseries", {
+        "machine": "m", "start": args["start"].isoformat(), "end": args["end"].isoformat(),
+        "variables": [{"uuid": "rate"}, {"uuid": "gate"}],
+    })
+    assert second.args == ("/v2/timeseries", {**first.args[1], "cursor": "continuation"})
+
+
+@pytest.mark.parametrize("metadata", [None, [], "bad", {},
+    {"next_cursor": ""}, {"next_cursor": 1}, {"next_cursor": False},
+    {"next_cursor": []}, {"next_cursor": {}}])
+def test_invalid_pagination_metadata(timeseries, metadata) -> None:
+    gateway, client, args = timeseries
+    client.post_json.return_value.body["metadata"] = metadata
+    with pytest.raises(ValueError, match="Invalid timeseries response.*next_cursor"):
+        gateway.get_material_samples(**args)
+    client.post_json.assert_called_once()
+
+
+@pytest.mark.parametrize("cursors", [["a", "a"], ["a", "b", "a"]])
+def test_repeated_pagination_cursor(timeseries, cursors) -> None:
+    gateway, client, args = timeseries
+    data = client.post_json.return_value.body["data"]
+    client.post_json.side_effect = [
+        Mock(body={"data": data, "metadata": {"next_cursor": cursor}}) for cursor in cursors
+    ]
+    with pytest.raises(ValueError, match="next_cursor has already been followed"):
+        gateway.get_material_samples(**args)
+    assert client.post_json.call_count == len(cursors)
+
+
+@pytest.mark.parametrize("data, error", [
+    ({"columns": ["time", "rate", "gate"], "records": [[]]}, "malformed record 0"),
+    ({"columns": ["time", "rate", "rate", "gate"], "records": []}, "required column"),
+    ({"columns": ["time", "rate"], "records": []}, "required column"),
+    ({"columns": None, "records": []}, "columns must be a list"),
+    ({"columns": ["time", "rate", "gate"], "records": None}, "records must be a list"),
+    ({"columns": ["time", "rate", "gate"],
+      "records": [["2026-09-03T01:00:00", 10, 2]]}, "timezone-aware"),
+    ({"columns": ["time", "rate", "gate"],
+      "records": [["2026-09-03T01:00:00Z", float("nan"), 2]]}, "finite"),
+    ({"columns": ["time", "rate", "gate"],
+      "records": [["2026-09-03T01:00:00Z", 10, True]]}, "numeric"),
+])
+def test_malformed_later_page(timeseries, data, error) -> None:
+    gateway, client, args = timeseries
+    first_body = client.post_json.return_value.body
+    first_body["metadata"] = {"next_cursor": "continuation"}
+    client.post_json.side_effect = [
+        Mock(body=first_body), Mock(body={"data": data, "metadata": {"next_cursor": None}}),
+    ]
+    with pytest.raises(ValueError, match=f"Invalid timeseries response on page 2:.*{error}"):
+        gateway.get_material_samples(**args)
+    assert client.post_json.call_count == 2