From 0e69efae9e7cfb414182c95964b043b1854b96e2 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 16 Jun 2026 00:45:09 +0000 Subject: [PATCH 01/10] query: add fieldFilters/sort/offset to list_runs Companion client change to pluto-server #490, which added server-side filtering, ordering, and offset pagination to GET /api/runs/list. - New FieldFilter dataclass ({source, key, dataType, operator, values}) with validation against the server's operator/source/dataType vocabulary; accepts FieldFilter objects or raw dicts. - Client.list_runs and the module-level list_runs gain field_filters (JSON-encoded, max 50 terms), sort (pass-through string), and offset (clamped to 0-100,000). Return type stays List[dict]. - Unit tests for the new params and FieldFilter validation; e2e tests for sort/offset/fieldFilters; a contract test that diffs the client's filter enums against the live OpenAPI FieldFilterTerm component (skips until the companion server change publishes that component). Co-Authored-By: Claude Opus 4.8 (1M context) --- pluto/query.py | 153 ++++++++++++++++++++++++++++++++++++++++- tests/test_contract.py | 89 ++++++++++++++++++++++++ tests/test_e2e.py | 57 +++++++++++++++ tests/test_query.py | 100 ++++++++++++++++++++++++++- 4 files changed, 397 insertions(+), 2 deletions(-) create mode 100644 tests/test_contract.py diff --git a/pluto/query.py b/pluto/query.py index 49277a2..d458474 100644 --- a/pluto/query.py +++ b/pluto/query.py @@ -18,10 +18,12 @@ runs = client.list_runs("my-project") """ +import json import logging import os import time import warnings +from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional, Union @@ -45,6 +47,100 @@ def __init__(self, message: str, status_code: Optional[int] = None): super().__init__(message) +# Field-filter vocabulary, kept in step with the server's zod schema for the +# ``/api/runs/list`` ``fieldFilters`` parameter. ``tests/test_contract.py`` +# asserts these match ``components.schemas.FieldFilterTerm`` in the live +# OpenAPI spec, so drift fails CI rather than surfacing as opaque HTTP 400s. +_FILTER_SOURCES = {'config', 'systemMetadata'} +_FILTER_DATATYPES = {'text', 'number', 'date', 'option'} +_FILTER_OPERATORS = { + 'contains', + 'does not contain', + 'is', + 'is not', + 'starts with', + 'ends with', + 'regex', + '>', + '<', + '>=', + '<=', + 'is between', + 'is any of', + 'is none of', +} + +# Server caps the filter set at 50 terms; reject early with a clear error. +_MAX_FILTER_TERMS = 50 +# Server clamps offset to this range (MAX_JSON_SORT_OFFSET). +_MAX_OFFSET = 100_000 + + +@dataclass +class FieldFilter: + """A single server-side filter term for :meth:`Client.list_runs`. + + Filters runs by an indexed ``config.*`` or ``systemMetadata.*`` field. + Multiple terms are AND-combined by the server. + + Args: + source: ``"config"`` or ``"systemMetadata"``. + key: Dotted field path (e.g. ``"checkpoint.r2_prefix"``). + dataType: One of ``"text"``, ``"number"``, ``"date"``, ``"option"``. + operator: One of the supported operators (e.g. ``"contains"``, + ``"is"``, ``">"``, ``"is between"``, ``"is any of"``). See + :data:`_FILTER_OPERATORS` for the full set. + values: Operand list. A scalar is coerced to a single-element list. + + Example:: + + FieldFilter("config", "lr", "number", ">", [0.001]) + FieldFilter("config", "model", "text", "contains", "gpt") + + Raises: + ValueError: If ``source``, ``dataType``, or ``operator`` is not a + recognised value. + """ + + source: str + key: str + dataType: str + operator: str + values: List[Any] = field(default_factory=list) + + def __post_init__(self) -> None: + if self.source not in _FILTER_SOURCES: + raise ValueError( + f'FieldFilter source must be one of {sorted(_FILTER_SOURCES)}, ' + f'got {self.source!r}' + ) + if self.dataType not in _FILTER_DATATYPES: + raise ValueError( + f'FieldFilter dataType must be one of {sorted(_FILTER_DATATYPES)}, ' + f'got {self.dataType!r}' + ) + if self.operator not in _FILTER_OPERATORS: + raise ValueError( + f'FieldFilter operator must be one of {sorted(_FILTER_OPERATORS)}, ' + f'got {self.operator!r}' + ) + # Accept a bare scalar for convenience (e.g. operator "is"). + if not isinstance(self.values, (list, tuple)): + self.values = [self.values] + else: + self.values = list(self.values) + + def to_dict(self) -> Dict[str, Any]: + """Serialize to the server's filter-term JSON shape.""" + return { + 'source': self.source, + 'key': self.key, + 'dataType': self.dataType, + 'operator': self.operator, + 'values': self.values, + } + + class Client: """HTTP client for reading data from the Pluto server. @@ -115,6 +211,9 @@ def list_runs( search: Optional[str] = None, tags: Optional[List[str]] = None, limit: int = 50, + field_filters: Optional[List[Union['FieldFilter', Dict[str, Any]]]] = None, + sort: Optional[str] = None, + offset: int = 0, ) -> List[Dict[str, Any]]: """List runs in a project. @@ -124,17 +223,58 @@ def list_runs( tags: Filter by tags (AND logic). Only runs matching *all* specified tags are returned. limit: Maximum number of runs to return (max 200). + field_filters: Server-side filters over ``config.*`` / + ``systemMetadata.*`` fields. A list of :class:`FieldFilter` + instances (or raw dicts in the same ``{source, key, dataType, + operator, values}`` shape). Terms are AND-combined. At most 50 + terms. + sort: Server-side ordering as ``[+|-]`` (``-`` = descending). + Accepts built-in columns (``createdAt``, ``updatedAt``, + ``name``, ``status`` and their snake_case aliases), + ``config..value`` / ``systemMetadata..value``, + ``summary_metrics.`` (LAST aggregation), and + ``heartbeat_at`` (last logged data point). Defaults to + ``createdAt desc`` server-side. + offset: Skip-based pagination offset (0–100,000). Combine with + ``limit`` to page; advance ``offset`` until a short page is + returned. Returns: List of run dicts with keys: ``id``, ``name``, ``displayId``, ``status``, ``tags``, ``config``, ``createdAt``, ``updatedAt``, ``url``. + + Raises: + ValueError: If more than 50 ``field_filters`` terms are given. """ params: Dict[str, Any] = {'projectName': project, 'limit': min(limit, 200)} if search is not None: params['search'] = search if tags is not None: params['tags'] = ','.join(tags) + if field_filters is not None: + if len(field_filters) > _MAX_FILTER_TERMS: + raise ValueError( + f'At most {_MAX_FILTER_TERMS} field_filters terms are ' + f'supported, got {len(field_filters)}' + ) + terms = [ + f.to_dict() if isinstance(f, FieldFilter) else f for f in field_filters + ] + params['fieldFilters'] = json.dumps(terms) + if sort is not None: + params['sort'] = sort + if offset: + clamped = max(0, min(offset, _MAX_OFFSET)) + if clamped != offset: + logger.debug( + '%s: offset %d clamped to %d (max %d)', + tag, + offset, + clamped, + _MAX_OFFSET, + ) + params['offset'] = clamped return self._get('/api/runs/list', params=params)['runs'] def get_run( @@ -549,9 +689,20 @@ def list_runs( search: Optional[str] = None, tags: Optional[List[str]] = None, limit: int = 50, + field_filters: Optional[List[Union['FieldFilter', Dict[str, Any]]]] = None, + sort: Optional[str] = None, + offset: int = 0, ) -> List[Dict[str, Any]]: """List runs in a project. See :meth:`Client.list_runs`.""" - return _get_client().list_runs(project, search=search, tags=tags, limit=limit) + return _get_client().list_runs( + project, + search=search, + tags=tags, + limit=limit, + field_filters=field_filters, + sort=sort, + offset=offset, + ) def get_run(project: str, run_id: Union[int, str]) -> Dict[str, Any]: diff --git a/tests/test_contract.py b/tests/test_contract.py new file mode 100644 index 0000000..a2a87db --- /dev/null +++ b/tests/test_contract.py @@ -0,0 +1,89 @@ +"""Contract test: keep the client's field-filter vocabulary in step with the +server's zod schema. + +The server's zod schema for the ``/api/runs/list`` ``fieldFilters`` parameter is +the source of truth. Rather than reimplement validation, we assert the client's +hardcoded enums match the schema the server publishes in its OpenAPI document +(served at ``{url_api}/api/openapi.json``), so any drift fails CI loudly instead +of surfacing as opaque HTTP 400s for users. + +This depends on the server exposing the inner filter-term schema as a structured +OpenAPI component (``components.schemas.FieldFilterTerm`` with real ``enum``s). +Until that companion server change lands, the spec carries the operators only as +prose in the parameter description, so this test SKIPS rather than fails. + +Network test: hits the live API spec endpoint (no auth required for the public +spec). Skips cleanly when the spec is unreachable, so offline/hermetic runs are +unaffected — matching the network-dependent style of ``tests/test_e2e.py``. +""" + +import httpx +import pytest + +from pluto.query import ( + _FILTER_DATATYPES, + _FILTER_OPERATORS, + _FILTER_SOURCES, + _resolve_url_api, +) + +# Component name the companion server PR registers in its OpenAPI document. +_COMPONENT = 'FieldFilterTerm' + + +def _fetch_openapi() -> dict: + url = f'{_resolve_url_api(None)}/api/openapi.json' + try: + resp = httpx.get(url, timeout=15, follow_redirects=True) + except httpx.HTTPError as exc: # pragma: no cover - network dependent + pytest.skip(f'Could not reach OpenAPI spec at {url}: {exc}') + if resp.status_code != 200: + pytest.skip(f'OpenAPI spec at {url} returned HTTP {resp.status_code}') + return resp.json() + + +def _term_schema() -> dict: + spec = _fetch_openapi() + schemas = spec.get('components', {}).get('schemas', {}) + if _COMPONENT not in schemas: + pytest.skip( + f'OpenAPI spec has no components.schemas.{_COMPONENT} yet; ' + 'pending the companion pluto-server change that surfaces the ' + 'field-filter zod schema as a structured component.' + ) + return schemas[_COMPONENT] + + +def _enum_for(term: dict, field: str) -> set: + prop = term.get('properties', {}).get(field, {}) + enum = prop.get('enum') + if enum is None: + pytest.skip(f'{_COMPONENT}.{field} has no enum in the OpenAPI spec') + return set(enum) + + +def test_filter_operators_match_server(): + server = _enum_for(_term_schema(), 'operator') + assert server == _FILTER_OPERATORS, ( + 'operator enum drift between client and server.\n' + f' server-only: {sorted(server - _FILTER_OPERATORS)}\n' + f' client-only: {sorted(_FILTER_OPERATORS - server)}' + ) + + +def test_filter_sources_match_server(): + server = _enum_for(_term_schema(), 'source') + assert server == _FILTER_SOURCES, ( + 'source enum drift between client and server.\n' + f' server-only: {sorted(server - _FILTER_SOURCES)}\n' + f' client-only: {sorted(_FILTER_SOURCES - server)}' + ) + + +def test_filter_datatypes_match_server(): + server = _enum_for(_term_schema(), 'dataType') + assert server == _FILTER_DATATYPES, ( + 'dataType enum drift between client and server.\n' + f' server-only: {sorted(server - _FILTER_DATATYPES)}\n' + f' client-only: {sorted(_FILTER_DATATYPES - server)}' + ) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 5ea1748..ca50f62 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -430,6 +430,63 @@ def test_e2e_list_runs_search(): assert run_id in found_ids, f'Run {run_id} (name={task_name}) not found via search' +def test_e2e_list_runs_sort_created_desc(): + """Verify sort='-createdAt' returns runs newest-first.""" + # Create two runs in sequence so their creation order is known. + run_a = pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), config={}) + id_a = run_a.settings._op_id + run_a.finish() + run_b = pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), config={}) + id_b = run_b.settings._op_id + run_b.finish() + + runs = pq.list_runs(TESTING_PROJECT_NAME, sort='-createdAt', limit=200) + ids = [r['id'] for r in runs] + assert id_a in ids and id_b in ids + # The later-created run (B) must appear before the earlier one (A). + assert ids.index(id_b) < ids.index(id_a), 'sort=-createdAt not newest-first' + + +def test_e2e_list_runs_offset_pagination(): + """Verify offset pagination skips runs.""" + # Ensure at least two runs exist. + pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), config={}).finish() + pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), config={}).finish() + + page1 = pq.list_runs(TESTING_PROJECT_NAME, sort='-createdAt', limit=1) + page2 = pq.list_runs(TESTING_PROJECT_NAME, sort='-createdAt', limit=1, offset=1) + assert len(page1) == 1 and len(page2) == 1 + assert page1[0]['id'] != page2[0]['id'], 'offset did not advance the page' + + +def test_e2e_list_runs_field_filter(): + """Verify fieldFilters filters runs by a config value, server-side.""" + marker = f'e2e-ff-{int(time.time())}' + run = pluto.init( + project=TESTING_PROJECT_NAME, + name=get_task_name(), + config={'e2e_filter_marker': marker}, + ) + run_id = run.settings._op_id + run.finish() + + def _query(): + runs = pq.list_runs( + TESTING_PROJECT_NAME, + field_filters=[ + pq.FieldFilter('config', 'e2e_filter_marker', 'text', 'is', [marker]) + ], + limit=200, + ) + ids = [r['id'] for r in runs] + return run_id in ids + + # Field values are indexed asynchronously; poll for eventual consistency. + assert _poll( + fn=_query, check=lambda found: found + ), f'Run {run_id} not found via fieldFilters on config.e2e_filter_marker' + + # --------------------------------------------------------------------------- # Histogram (structured data) # --------------------------------------------------------------------------- diff --git a/tests/test_query.py b/tests/test_query.py index dec315d..1639ea9 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -207,6 +207,98 @@ def test_limit_capped_at_200(self, client, mock_response): call_args = client._client.get.call_args assert call_args[1]['params']['limit'] == 200 + def test_sort(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + client.list_runs('proj', sort='-createdAt') + call_args = client._client.get.call_args + assert call_args[1]['params']['sort'] == '-createdAt' + + def test_offset(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + client.list_runs('proj', offset=100) + call_args = client._client.get.call_args + assert call_args[1]['params']['offset'] == 100 + + def test_offset_zero_omitted(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + client.list_runs('proj') + call_args = client._client.get.call_args + assert 'offset' not in call_args[1]['params'] + + def test_offset_clamped(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + client.list_runs('proj', offset=10**9) + call_args = client._client.get.call_args + assert call_args[1]['params']['offset'] == 100_000 + + def test_field_filters_objects(self, client, mock_response): + from pluto.query import FieldFilter + + client._client.get.return_value = mock_response(200, {'runs': []}) + client.list_runs( + 'proj', + field_filters=[FieldFilter('config', 'lr', 'number', '>', [0.001])], + ) + call_args = client._client.get.call_args + sent = json.loads(call_args[1]['params']['fieldFilters']) + assert sent == [ + { + 'source': 'config', + 'key': 'lr', + 'dataType': 'number', + 'operator': '>', + 'values': [0.001], + } + ] + + def test_field_filters_raw_dicts(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + term = { + 'source': 'systemMetadata', + 'key': 'gpu', + 'dataType': 'text', + 'operator': 'contains', + 'values': ['a100'], + } + client.list_runs('proj', field_filters=[term]) + call_args = client._client.get.call_args + assert json.loads(call_args[1]['params']['fieldFilters']) == [term] + + def test_field_filters_too_many_raises(self, client, mock_response): + from pluto.query import FieldFilter + + client._client.get.return_value = mock_response(200, {'runs': []}) + terms = [FieldFilter('config', f'k{i}', 'text', 'is', ['x']) for i in range(51)] + with pytest.raises(ValueError, match='At most 50'): + client.list_runs('proj', field_filters=terms) + + +class TestFieldFilter: + def test_scalar_values_coerced_to_list(self): + from pluto.query import FieldFilter + + f = FieldFilter('config', 'model', 'text', 'is', 'gpt') + assert f.values == ['gpt'] + assert f.to_dict()['values'] == ['gpt'] + + def test_bad_source_raises(self): + from pluto.query import FieldFilter + + with pytest.raises(ValueError, match='source'): + FieldFilter('cfg', 'lr', 'number', '>', [1]) + + def test_bad_datatype_raises(self): + from pluto.query import FieldFilter + + with pytest.raises(ValueError, match='dataType'): + FieldFilter('config', 'lr', 'float', '>', [1]) + + def test_bad_operator_raises(self): + from pluto.query import FieldFilter + + with pytest.raises(ValueError, match='operator'): + FieldFilter('config', 'lr', 'number', 'gt', [1]) + # --------------------------------------------------------------------------- # get_run @@ -520,7 +612,13 @@ def test_list_runs_creates_default_client(self, monkeypatch, mock_response): assert result == [{'id': 1}] mock_client_instance.list_runs.assert_called_once_with( - 'my-project', search=None, tags=None, limit=50 + 'my-project', + search=None, + tags=None, + limit=50, + field_filters=None, + sort=None, + offset=0, ) # Clean up From 2d192adb253a21451d56c120f027fe476aae5a6b Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 16 Jun 2026 01:04:22 +0000 Subject: [PATCH 02/10] =?UTF-8?q?query:=20address=20review=20=E2=80=94=20d?= =?UTF-8?q?on't=20send=20clamped-to-0=20offset;=20skip=20contract=20test?= =?UTF-8?q?=20on=20non-JSON=20spec?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - list_runs: a negative offset clamps to 0 and is now omitted rather than sent as a redundant offset=0 param. - test_contract: skip gracefully if /api/openapi.json returns a non-JSON body. Co-Authored-By: Claude Opus 4.8 (1M context) --- pluto/query.py | 4 +++- tests/test_contract.py | 5 ++++- tests/test_query.py | 6 ++++++ 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/pluto/query.py b/pluto/query.py index d458474..b342d5e 100644 --- a/pluto/query.py +++ b/pluto/query.py @@ -274,7 +274,9 @@ def list_runs( clamped, _MAX_OFFSET, ) - params['offset'] = clamped + # A negative offset clamps to 0; don't send a redundant offset=0. + if clamped: + params['offset'] = clamped return self._get('/api/runs/list', params=params)['runs'] def get_run( diff --git a/tests/test_contract.py b/tests/test_contract.py index a2a87db..da9bc24 100644 --- a/tests/test_contract.py +++ b/tests/test_contract.py @@ -39,7 +39,10 @@ def _fetch_openapi() -> dict: pytest.skip(f'Could not reach OpenAPI spec at {url}: {exc}') if resp.status_code != 200: pytest.skip(f'OpenAPI spec at {url} returned HTTP {resp.status_code}') - return resp.json() + try: + return resp.json() + except ValueError as exc: # invalid/non-JSON body (e.g. an HTML error page) + pytest.skip(f'OpenAPI spec at {url} returned non-JSON body: {exc}') def _term_schema() -> dict: diff --git a/tests/test_query.py b/tests/test_query.py index 1639ea9..6af2337 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -231,6 +231,12 @@ def test_offset_clamped(self, client, mock_response): call_args = client._client.get.call_args assert call_args[1]['params']['offset'] == 100_000 + def test_negative_offset_omitted(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + client.list_runs('proj', offset=-5) + call_args = client._client.get.call_args + assert 'offset' not in call_args[1]['params'] + def test_field_filters_objects(self, client, mock_response): from pluto.query import FieldFilter From 731e052036e51d89dcc161bb41f25dee76edcdfd Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 16 Jun 2026 18:29:06 +0000 Subject: [PATCH 03/10] ci: run schema contract test against prod and staging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a reusable contract-test workflow (workflow_call, parameterised by API base URL) with two thin callers — prod (pluto-api.trainy.ai) and staging (pluto-api-dev.trainy.ai) — so the OpenAPI FieldFilterTerm schema check runs against both on every PR/push. Staging deploys first, so client/server enum drift is caught before it reaches prod. Also send Authorization: Bearer $PLUTO_API_KEY when fetching the spec, so the check still runs if staging gates /api/openapi.json behind auth (harmless on the unauthenticated prod endpoint). Both jobs pass-with-skips until the server publishes the FieldFilterTerm component; they self-activate once it appears. Co-Authored-By: Claude Opus 4.8 (1M context) --- .github/workflows/contract-test-prod.yml | 23 +++++++++ .github/workflows/contract-test-staging.yml | 25 +++++++++ .github/workflows/contract-test.yml | 57 +++++++++++++++++++++ tests/test_contract.py | 11 +++- 4 files changed, 115 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/contract-test-prod.yml create mode 100644 .github/workflows/contract-test-staging.yml create mode 100644 .github/workflows/contract-test.yml diff --git a/.github/workflows/contract-test-prod.yml b/.github/workflows/contract-test-prod.yml new file mode 100644 index 0000000..e66955a --- /dev/null +++ b/.github/workflows/contract-test-prod.yml @@ -0,0 +1,23 @@ +name: contract-test (prod) + +# Runs the OpenAPI schema contract test against production. + +on: + push: + branches: + - main + - 'releases/**' + pull_request: + branches: + - main + - 'releases/**' + workflow_dispatch: + +permissions: {} + +jobs: + prod: + uses: ./.github/workflows/contract-test.yml + with: + url_api: https://pluto-api.trainy.ai + secrets: inherit diff --git a/.github/workflows/contract-test-staging.yml b/.github/workflows/contract-test-staging.yml new file mode 100644 index 0000000..4aaad4e --- /dev/null +++ b/.github/workflows/contract-test-staging.yml @@ -0,0 +1,25 @@ +name: contract-test (staging) + +# Runs the OpenAPI schema contract test against staging/dev, where the server +# side typically deploys first — so client/server enum drift is caught before +# it reaches production. + +on: + push: + branches: + - main + - 'releases/**' + pull_request: + branches: + - main + - 'releases/**' + workflow_dispatch: + +permissions: {} + +jobs: + staging: + uses: ./.github/workflows/contract-test.yml + with: + url_api: https://pluto-api-dev.trainy.ai + secrets: inherit diff --git a/.github/workflows/contract-test.yml b/.github/workflows/contract-test.yml new file mode 100644 index 0000000..8da96c3 --- /dev/null +++ b/.github/workflows/contract-test.yml @@ -0,0 +1,57 @@ +name: contract-test + +# Reusable workflow: run the OpenAPI schema contract test (tests/test_contract.py) +# against a given Pluto API host. Callers (contract-test-prod.yml, +# contract-test-staging.yml) supply the base URL so the two environments share +# one identical job definition and can never drift. + +on: + workflow_call: + inputs: + url_api: + description: Pluto API base URL (e.g. https://pluto-api.trainy.ai) + required: true + type: string + +permissions: {} + +jobs: + contract: + # Reuse the "integration" environment so secrets.MLOP_API_TOKEN resolves. + environment: integration + permissions: + contents: read + runs-on: ubicloud-standard-4 + strategy: + matrix: + python-version: ["3.11"] + poetry-version: ["2.1.1"] + steps: + - name: Checkout code + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + + - name: Set up Python + uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0 + with: + python-version: ${{ matrix.python-version }} + + - name: Install Poetry + run: pip install poetry==${{ matrix.poetry-version }} + + - name: Install deps + run: | + set -euox pipefail + # Clear stale cached venv that may be missing dev dependencies. + rm -rf .venv + # `--extras full` is required even for this light job: importing + # pluto.query pulls in pluto/__init__.py -> pluto/file.py -> soundfile. + poetry install --with dev --extras full + pip install -e ".[dev,full]" + + - name: Run contract test + env: + PLUTO_URL_API: ${{ inputs.url_api }} + PLUTO_API_KEY: ${{ secrets.MLOP_API_TOKEN }} + run: | + set -euox pipefail + poetry run pytest tests/test_contract.py -rs -v diff --git a/tests/test_contract.py b/tests/test_contract.py index da9bc24..c5ad75b 100644 --- a/tests/test_contract.py +++ b/tests/test_contract.py @@ -17,6 +17,8 @@ unaffected — matching the network-dependent style of ``tests/test_e2e.py``. """ +import os + import httpx import pytest @@ -33,8 +35,15 @@ def _fetch_openapi() -> dict: url = f'{_resolve_url_api(None)}/api/openapi.json' + # Prod serves the spec unauthenticated; staging/dev may gate it. Send the + # bearer token when present so the check actually runs there. Harmless on an + # unauthenticated endpoint. + headers = {} + token = os.environ.get('PLUTO_API_KEY') + if token: + headers['Authorization'] = f'Bearer {token}' try: - resp = httpx.get(url, timeout=15, follow_redirects=True) + resp = httpx.get(url, timeout=15, follow_redirects=True, headers=headers) except httpx.HTTPError as exc: # pragma: no cover - network dependent pytest.skip(f'Could not reach OpenAPI spec at {url}: {exc}') if resp.status_code != 200: From ac6e4864cb144e96bb04f40985fa8eb0ccf5bf1b Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 16 Jun 2026 18:31:37 +0000 Subject: [PATCH 04/10] ci: grant contents:read to contract-test callers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reusable-workflow callers with permissions:{} caused startup_failure — a called workflow cannot receive more GITHUB_TOKEN permission than its caller, and the reusable job requests contents:read. Grant it on the callers. Co-Authored-By: Claude Opus 4.8 (1M context) --- .github/workflows/contract-test-prod.yml | 5 ++++- .github/workflows/contract-test-staging.yml | 5 ++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/.github/workflows/contract-test-prod.yml b/.github/workflows/contract-test-prod.yml index e66955a..05db8fb 100644 --- a/.github/workflows/contract-test-prod.yml +++ b/.github/workflows/contract-test-prod.yml @@ -13,7 +13,10 @@ on: - 'releases/**' workflow_dispatch: -permissions: {} +# Must grant at least what the called workflow's job requests (contents: read); +# a called workflow cannot receive more token permission than its caller. +permissions: + contents: read jobs: prod: diff --git a/.github/workflows/contract-test-staging.yml b/.github/workflows/contract-test-staging.yml index 4aaad4e..59f2caf 100644 --- a/.github/workflows/contract-test-staging.yml +++ b/.github/workflows/contract-test-staging.yml @@ -15,7 +15,10 @@ on: - 'releases/**' workflow_dispatch: -permissions: {} +# Must grant at least what the called workflow's job requests (contents: read); +# a called workflow cannot receive more token permission than its caller. +permissions: + contents: read jobs: staging: From b08644174d62422f0a8c1cf8a0a5fa8e9d558f1c Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 16 Jun 2026 18:42:09 +0000 Subject: [PATCH 05/10] test(e2e): make sort/offset list_runs tests race-free test_e2e_list_runs_offset_pagination flaked on the 3.12 matrix leg: against the shared, concurrently-written testing-ci project, a new run was inserted between the two list_runs calls, so sort=-createdAt offset=1 shifted back onto the run offset=0 had just returned (assert 164647 != 164647). Scope both the sort and offset assertions to a controlled two-run set sharing a unique tag (poll until both are listable; 1.1s gap for distinct createdAt), so concurrent writes can't move the window. Mirrors the approach already used by test_e2e_list_runs_field_filter. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_e2e.py | 59 ++++++++++++++++++++++++++++++++--------------- 1 file changed, 41 insertions(+), 18 deletions(-) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index ca50f62..64d1842 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -13,6 +13,7 @@ import importlib.util import io import time +import uuid from typing import Callable, List, TypeVar import httpx @@ -430,33 +431,55 @@ def test_e2e_list_runs_search(): assert run_id in found_ids, f'Run {run_id} (name={task_name}) not found via search' -def test_e2e_list_runs_sort_created_desc(): - """Verify sort='-createdAt' returns runs newest-first.""" - # Create two runs in sequence so their creation order is known. - run_a = pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), config={}) +def _two_tagged_runs() -> tuple: + """Create two finished runs (A older than B) sharing a unique tag. + + Returns ``(tag, id_a, id_b)`` once both are listable under the tag. The + unique tag scopes subsequent list queries to exactly these two runs, so + concurrent writes to the shared ``testing-ci`` project (parallel matrix + legs, ``-n auto`` workers) can't shift the sort/pagination window. A >1s + gap guarantees distinct ``createdAt`` even at second precision. + """ + tag = f'e2e-page-{uuid.uuid4().hex[:12]}' + run_a = pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), tags=[tag]) id_a = run_a.settings._op_id run_a.finish() - run_b = pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), config={}) + time.sleep(1.1) + run_b = pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), tags=[tag]) id_b = run_b.settings._op_id run_b.finish() - runs = pq.list_runs(TESTING_PROJECT_NAME, sort='-createdAt', limit=200) + # Tags sync asynchronously; poll until both runs are listable under the tag. + def _both_listed() -> bool: + ids = { + r['id'] for r in pq.list_runs(TESTING_PROJECT_NAME, tags=[tag], limit=200) + } + return {id_a, id_b} <= ids + + assert _poll( + fn=_both_listed, check=lambda ok: ok + ), f'tagged runs {id_a},{id_b} not both listable under {tag}' + return tag, id_a, id_b + + +def test_e2e_list_runs_sort_created_desc(): + """Verify sort='-createdAt' orders newest-first within a controlled set.""" + tag, id_a, id_b = _two_tagged_runs() + runs = pq.list_runs(TESTING_PROJECT_NAME, tags=[tag], sort='-createdAt', limit=200) ids = [r['id'] for r in runs] - assert id_a in ids and id_b in ids - # The later-created run (B) must appear before the earlier one (A). - assert ids.index(id_b) < ids.index(id_a), 'sort=-createdAt not newest-first' + # Exactly our two runs match the tag; B (later) must come before A. + assert ids == [id_b, id_a], f'sort=-createdAt not newest-first: {ids}' def test_e2e_list_runs_offset_pagination(): - """Verify offset pagination skips runs.""" - # Ensure at least two runs exist. - pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), config={}).finish() - pluto.init(project=TESTING_PROJECT_NAME, name=get_task_name(), config={}).finish() - - page1 = pq.list_runs(TESTING_PROJECT_NAME, sort='-createdAt', limit=1) - page2 = pq.list_runs(TESTING_PROJECT_NAME, sort='-createdAt', limit=1, offset=1) - assert len(page1) == 1 and len(page2) == 1 - assert page1[0]['id'] != page2[0]['id'], 'offset did not advance the page' + """Verify offset advances the page within a controlled, stable set.""" + tag, id_a, id_b = _two_tagged_runs() + page1 = pq.list_runs(TESTING_PROJECT_NAME, tags=[tag], sort='-createdAt', limit=1) + page2 = pq.list_runs( + TESTING_PROJECT_NAME, tags=[tag], sort='-createdAt', limit=1, offset=1 + ) + assert [r['id'] for r in page1] == [id_b], 'page 1 should be the newest run' + assert [r['id'] for r in page2] == [id_a], 'offset=1 should skip to the older run' def test_e2e_list_runs_field_filter(): From fbbc9e2a08080b481e0d788c36de2afa6d62d7f5 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Mon, 22 Jun 2026 23:56:35 +0000 Subject: [PATCH 06/10] query: support the full field-filter operator set Expand FieldFilter beyond the documented 14 operators to the complete vocabulary the server's run_field_values builder accepts (26 ops). Operators are now scoped to the dataType they apply to (the server dispatches on dataType): text gains equals; number gains is greater than/is less than (+ >=/<= phrase forms) and is not between; date gains is before/is on or before/is after/is on or after/is between/is not between; plus exists/not exists for any dataType. - _FILTER_OPERATORS_BY_DATATYPE map + _FILTER_OPERATORS_COMMON; FieldFilter validates the operator against its dataType (clearer error, catches e.g. "contains" on a number field). The flat _FILTER_OPERATORS union (26) is what the contract test compares to the server's published FieldFilterTerm enum. - Verified byte-identical to the server companion PR's FIELD_FILTER_OPERATORS. Co-Authored-By: Claude Opus 4.8 (1M context) --- pluto/query.py | 100 ++++++++++++++++++++++++++++++++------------ tests/test_query.py | 50 ++++++++++++++++++++++ 2 files changed, 124 insertions(+), 26 deletions(-) diff --git a/pluto/query.py b/pluto/query.py index b342d5e..12e9a93 100644 --- a/pluto/query.py +++ b/pluto/query.py @@ -48,27 +48,63 @@ def __init__(self, message: str, status_code: Optional[int] = None): # Field-filter vocabulary, kept in step with the server's zod schema for the -# ``/api/runs/list`` ``fieldFilters`` parameter. ``tests/test_contract.py`` -# asserts these match ``components.schemas.FieldFilterTerm`` in the live -# OpenAPI spec, so drift fails CI rather than surfacing as opaque HTTP 400s. +# ``/api/runs/list`` ``fieldFilters`` parameter (pluto-server +# ``buildValueCondition``). Operators are scoped to the dataType they apply to — +# the server dispatches on dataType — so e.g. ``contains`` is text-only and ``>`` +# is number-only. ``exists``/``not exists`` work for any dataType. _FILTER_SOURCES = {'config', 'systemMetadata'} _FILTER_DATATYPES = {'text', 'number', 'date', 'option'} -_FILTER_OPERATORS = { - 'contains', - 'does not contain', - 'is', - 'is not', - 'starts with', - 'ends with', - 'regex', - '>', - '<', - '>=', - '<=', - 'is between', - 'is any of', - 'is none of', +_FILTER_OPERATORS_COMMON = {'exists', 'not exists'} +_FILTER_OPERATORS_BY_DATATYPE = { + 'text': { + 'contains', + 'does not contain', + 'equals', + 'is', + 'is not', + 'starts with', + 'ends with', + 'regex', + }, + 'number': { + 'is', + 'is not', + 'is greater than', + '>', + 'is less than', + '<', + 'is greater than or equal to', + '>=', + 'is less than or equal to', + '<=', + 'is between', + 'is not between', + }, + 'date': { + 'is before', + 'is on or before', + 'is after', + 'is on or after', + 'is between', + 'is not between', + }, + 'option': {'is', 'is not', 'is any of', 'is none of'}, } +# Flat union — the single operator enum the server publishes in OpenAPI. +# ``tests/test_contract.py`` asserts this equals +# ``components.schemas.FieldFilterTerm.properties.operator.enum`` in the live +# spec, so drift fails CI rather than surfacing as opaque HTTP 400s. +_FILTER_OPERATORS = _FILTER_OPERATORS_COMMON.union( + *_FILTER_OPERATORS_BY_DATATYPE.values() +) + + +def _allowed_operators(data_type: str) -> set: + """Operators valid for *data_type* (its dataType-specific set + common).""" + return ( + _FILTER_OPERATORS_BY_DATATYPE.get(data_type, set()) | _FILTER_OPERATORS_COMMON + ) + # Server caps the filter set at 50 terms; reject early with a clear error. _MAX_FILTER_TERMS = 50 @@ -87,19 +123,30 @@ class FieldFilter: source: ``"config"`` or ``"systemMetadata"``. key: Dotted field path (e.g. ``"checkpoint.r2_prefix"``). dataType: One of ``"text"``, ``"number"``, ``"date"``, ``"option"``. - operator: One of the supported operators (e.g. ``"contains"``, - ``"is"``, ``">"``, ``"is between"``, ``"is any of"``). See - :data:`_FILTER_OPERATORS` for the full set. + operator: An operator valid for *dataType* (the server dispatches on + dataType). text: ``contains``, ``does not contain``, ``equals``, + ``is``, ``is not``, ``starts with``, ``ends with``, ``regex``; + number: ``is``, ``is not``, ``is greater than``/``>``, + ``is less than``/``<``, ``is greater than or equal to``/``>=``, + ``is less than or equal to``/``<=``, ``is between``, + ``is not between``; date: ``is before``, ``is on or before``, + ``is after``, ``is on or after``, ``is between``, ``is not between``; + option: ``is``, ``is not``, ``is any of``, ``is none of``. Any + dataType also accepts ``exists`` / ``not exists``. values: Operand list. A scalar is coerced to a single-element list. + ``is between`` takes two values; ``exists`` / ``not exists`` take + none. Example:: FieldFilter("config", "lr", "number", ">", [0.001]) + FieldFilter("config", "lr", "number", "is between", [0.1, 0.9]) FieldFilter("config", "model", "text", "contains", "gpt") + FieldFilter("config", "checkpoint", "text", "exists") Raises: - ValueError: If ``source``, ``dataType``, or ``operator`` is not a - recognised value. + ValueError: If ``source`` or ``dataType`` is unrecognised, or + ``operator`` is not valid for ``dataType``. """ source: str @@ -119,10 +166,11 @@ def __post_init__(self) -> None: f'FieldFilter dataType must be one of {sorted(_FILTER_DATATYPES)}, ' f'got {self.dataType!r}' ) - if self.operator not in _FILTER_OPERATORS: + allowed = _allowed_operators(self.dataType) + if self.operator not in allowed: raise ValueError( - f'FieldFilter operator must be one of {sorted(_FILTER_OPERATORS)}, ' - f'got {self.operator!r}' + f'FieldFilter operator {self.operator!r} is not valid for ' + f'dataType {self.dataType!r}; expected one of {sorted(allowed)}' ) # Accept a bare scalar for convenience (e.g. operator "is"). if not isinstance(self.values, (list, tuple)): diff --git a/tests/test_query.py b/tests/test_query.py index 6af2337..3f0a731 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -305,6 +305,56 @@ def test_bad_operator_raises(self): with pytest.raises(ValueError, match='operator'): FieldFilter('config', 'lr', 'number', 'gt', [1]) + @pytest.mark.parametrize( + 'source,key,dtype,op,vals', + [ + ('config', 'lr', 'number', 'is between', [0.1, 0.9]), + ('config', 'lr', 'number', 'is greater than', [0.1]), + ('config', 'ts', 'date', 'is before', ['2026-01-01']), + ('config', 'ts', 'date', 'is on or after', ['2026-01-01']), + ('config', 'name', 'text', 'starts with', ['gpt']), + ('systemMetadata', 'gpu', 'option', 'is any of', ['a100', 'h100']), + ('config', 'checkpoint', 'text', 'exists', []), + ('config', 'lr', 'number', 'not exists', []), + ], + ) + def test_valid_operators_accepted(self, source, key, dtype, op, vals): + from pluto.query import FieldFilter + + f = FieldFilter(source, key, dtype, op, vals) + assert f.to_dict()['operator'] == op + + @pytest.mark.parametrize( + 'dtype,op', + [ + ('number', 'contains'), # text-only operator on number + ('date', 'regex'), # text-only operator on date + ('number', 'is any of'), # option-only operator on number + ('option', '>'), # number-only operator on option + ('text', 'is between'), # number-only operator on text + ], + ) + def test_operator_rejected_for_wrong_datatype(self, dtype, op): + from pluto.query import FieldFilter + + with pytest.raises(ValueError, match='not valid for dataType'): + FieldFilter('config', 'k', dtype, op, ['x']) + + def test_flat_operator_set_is_union(self): + # The flat set the contract test compares against must be exactly the + # union of the per-dataType sets plus the common operators. + from pluto.query import ( + _FILTER_OPERATORS, + _FILTER_OPERATORS_BY_DATATYPE, + _FILTER_OPERATORS_COMMON, + ) + + expected = set(_FILTER_OPERATORS_COMMON) + for ops in _FILTER_OPERATORS_BY_DATATYPE.values(): + expected |= ops + assert _FILTER_OPERATORS == expected + assert len(_FILTER_OPERATORS) == 26 + # --------------------------------------------------------------------------- # get_run From 5ebdf713ee9058a0d211e8669b9fa8e7d9581d65 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 23 Jun 2026 21:49:16 +0000 Subject: [PATCH 07/10] query: regen API docs + validate field_filters item types - Merge main (picks up the api-docs autodoc workflow); regenerate docs-api/query.mdx so the `check` job passes after the FieldFilter/list_runs docstring changes. - Address review: list_runs now raises TypeError for a field_filters item that is neither a FieldFilter nor a dict (previously serialized through to a server 400). Co-Authored-By: Claude Opus 4.8 (1M context) --- docs-api/query.mdx | 13 +++++++++++++ pluto/query.py | 14 +++++++++++--- tests/test_query.py | 5 +++++ 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/docs-api/query.mdx b/docs-api/query.mdx index 41a3d35..d4ca614 100644 --- a/docs-api/query.mdx +++ b/docs-api/query.mdx @@ -44,6 +44,9 @@ list_runs( search: Optional[str] = None, tags: Optional[List[str]] = None, limit: int = 50, + field_filters: Optional[List[Union[FieldFilter, Dict[str, Any]]]] = None, + sort: Optional[str] = None, + offset: int = 0, ) -> List[Dict[str, Any]] ``` @@ -55,9 +58,16 @@ List runs in a project. | `search` | `Optional[str]` | Full-text search on run name. | | `tags` | `Optional[List[str]]` | Filter by tags (AND logic). Only runs matching *all* specified tags are returned. | | `limit` | `int` | Maximum number of runs to return (max 200). | +| `field_filters` | `Optional[List[Union[FieldFilter, Dict[str, Any]]]]` | Server-side filters over `config.*` / `systemMetadata.*` fields. A list of `FieldFilter` instances (or raw dicts in the same `{source, key, dataType, operator, values}` shape). Terms are AND-combined. At most 50 terms. | +| `sort` | `Optional[str]` | Server-side ordering as `[+\|-]` (`-` = descending). Accepts built-in columns (`createdAt`, `updatedAt`, `name`, `status` and their snake_case aliases), `config..value` / `systemMetadata..value`, `summary_metrics.` (LAST aggregation), and `heartbeat_at` (last logged data point). Defaults to `createdAt desc` server-side. | +| `offset` | `int` | Skip-based pagination offset (0–100,000). Combine with `limit` to page; advance `offset` until a short page is returned. | **Returns:** `List[Dict[str, Any]]` — List of run dicts with keys: `id`, `name`, `displayId`, `status`, `tags`, `config`, `createdAt`, `updatedAt`, `url`. +**Raises:** + +- `ValueError` — If more than 50 `field_filters` terms are given. + #### `Client.get_run` ```python @@ -275,6 +285,9 @@ list_runs( search: Optional[str] = None, tags: Optional[List[str]] = None, limit: int = 50, + field_filters: Optional[List[Union[FieldFilter, Dict[str, Any]]]] = None, + sort: Optional[str] = None, + offset: int = 0, ) -> List[Dict[str, Any]] ``` diff --git a/pluto/query.py b/pluto/query.py index 12e9a93..32b7ecb 100644 --- a/pluto/query.py +++ b/pluto/query.py @@ -306,9 +306,17 @@ def list_runs( f'At most {_MAX_FILTER_TERMS} field_filters terms are ' f'supported, got {len(field_filters)}' ) - terms = [ - f.to_dict() if isinstance(f, FieldFilter) else f for f in field_filters - ] + terms = [] + for f in field_filters: + if isinstance(f, FieldFilter): + terms.append(f.to_dict()) + elif isinstance(f, dict): + terms.append(f) + else: + raise TypeError( + 'each field_filters item must be a FieldFilter or dict, ' + f'got {type(f).__name__}' + ) params['fieldFilters'] = json.dumps(terms) if sort is not None: params['sort'] = sort diff --git a/tests/test_query.py b/tests/test_query.py index 3f0a731..1c2b7bf 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -270,6 +270,11 @@ def test_field_filters_raw_dicts(self, client, mock_response): call_args = client._client.get.call_args assert json.loads(call_args[1]['params']['fieldFilters']) == [term] + def test_field_filters_invalid_item_type_raises(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + with pytest.raises(TypeError, match='FieldFilter or dict'): + client.list_runs('proj', field_filters=['not-a-filter']) + def test_field_filters_too_many_raises(self, client, mock_response): from pluto.query import FieldFilter From ad526f962d0bcfead0af34f41f70a2ce74741a17 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 23 Jun 2026 22:53:56 +0000 Subject: [PATCH 08/10] query: add status + heartbeat_after/before filters to list_runs Companion to server PR #512. Lets clients filter runs by lifecycle status and last-activity (heartbeat) time. The Linum spot-retry query becomes one call: list_runs(project, status=['RUNNING','FAILED','TERMINATED','CANCELLED'], heartbeat_before=cutoff_iso) # stale, non-completed => retry - status: list[str] validated against the 5 RunStatus values (ValueError on bad), forwarded as the comma-separated `status` param. - heartbeat_after / heartbeat_before: ISO-8601, forwarded as heartbeatAfter/ heartbeatBefore (server requires projectName for these). - Docstring documents the retry recipe; regenerated docs-api/query.mdx. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs-api/query.mdx | 11 ++++++++++- pluto/query.py | 46 ++++++++++++++++++++++++++++++++++++++++++++- tests/test_query.py | 25 ++++++++++++++++++++++++ 3 files changed, 80 insertions(+), 2 deletions(-) diff --git a/docs-api/query.mdx b/docs-api/query.mdx index d4ca614..d8fc424 100644 --- a/docs-api/query.mdx +++ b/docs-api/query.mdx @@ -47,6 +47,9 @@ list_runs( field_filters: Optional[List[Union[FieldFilter, Dict[str, Any]]]] = None, sort: Optional[str] = None, offset: int = 0, + status: Optional[List[str]] = None, + heartbeat_after: Optional[str] = None, + heartbeat_before: Optional[str] = None, ) -> List[Dict[str, Any]] ``` @@ -61,12 +64,15 @@ List runs in a project. | `field_filters` | `Optional[List[Union[FieldFilter, Dict[str, Any]]]]` | Server-side filters over `config.*` / `systemMetadata.*` fields. A list of `FieldFilter` instances (or raw dicts in the same `{source, key, dataType, operator, values}` shape). Terms are AND-combined. At most 50 terms. | | `sort` | `Optional[str]` | Server-side ordering as `[+\|-]` (`-` = descending). Accepts built-in columns (`createdAt`, `updatedAt`, `name`, `status` and their snake_case aliases), `config..value` / `systemMetadata..value`, `summary_metrics.` (LAST aggregation), and `heartbeat_at` (last logged data point). Defaults to `createdAt desc` server-side. | | `offset` | `int` | Skip-based pagination offset (0–100,000). Combine with `limit` to page; advance `offset` until a short page is returned. | +| `status` | `Optional[List[str]]` | Keep only runs whose status is in this list (OR within the list). Allowed: `RUNNING`, `COMPLETED`, `FAILED`, `TERMINATED`, `CANCELLED`. | +| `heartbeat_after` | `Optional[str]` | ISO-8601 timestamp; keep only runs whose last logged-data time (heartbeat) is at or after this. Requires `project`. | +| `heartbeat_before` | `Optional[str]` | ISO-8601 timestamp; keep only runs whose heartbeat is at or before this (i.e. *stale*). Requires `project`. | **Returns:** `List[Dict[str, Any]]` — List of run dicts with keys: `id`, `name`, `displayId`, `status`, `tags`, `config`, `createdAt`, `updatedAt`, `url`. **Raises:** -- `ValueError` — If more than 50 `field_filters` terms are given. +- `ValueError` — If more than 50 `field_filters` terms are given, or a `status` value is not a recognised run status. #### `Client.get_run` @@ -288,6 +294,9 @@ list_runs( field_filters: Optional[List[Union[FieldFilter, Dict[str, Any]]]] = None, sort: Optional[str] = None, offset: int = 0, + status: Optional[List[str]] = None, + heartbeat_after: Optional[str] = None, + heartbeat_before: Optional[str] = None, ) -> List[Dict[str, Any]] ``` diff --git a/pluto/query.py b/pluto/query.py index 32b7ecb..b7d0267 100644 --- a/pluto/query.py +++ b/pluto/query.py @@ -110,6 +110,8 @@ def _allowed_operators(data_type: str) -> set: _MAX_FILTER_TERMS = 50 # Server clamps offset to this range (MAX_JSON_SORT_OFFSET). _MAX_OFFSET = 100_000 +# Run lifecycle statuses accepted by the server's ``status`` filter. +_RUN_STATUSES = {'RUNNING', 'COMPLETED', 'FAILED', 'TERMINATED', 'CANCELLED'} @dataclass @@ -262,6 +264,9 @@ def list_runs( field_filters: Optional[List[Union['FieldFilter', Dict[str, Any]]]] = None, sort: Optional[str] = None, offset: int = 0, + status: Optional[List[str]] = None, + heartbeat_after: Optional[str] = None, + heartbeat_before: Optional[str] = None, ) -> List[Dict[str, Any]]: """List runs in a project. @@ -286,6 +291,25 @@ def list_runs( offset: Skip-based pagination offset (0–100,000). Combine with ``limit`` to page; advance ``offset`` until a short page is returned. + status: Keep only runs whose status is in this list (OR within the + list). Allowed: ``RUNNING``, ``COMPLETED``, ``FAILED``, + ``TERMINATED``, ``CANCELLED``. + heartbeat_after: ISO-8601 timestamp; keep only runs whose last + logged-data time (heartbeat) is at or after this. Requires + ``project``. + heartbeat_before: ISO-8601 timestamp; keep only runs whose heartbeat + is at or before this (i.e. *stale*). Requires ``project``. + + All filters AND together. To find interrupted jobs to retry (the wandb + "stale spot run" pattern), select non-completed runs that haven't + reported data recently:: + + list_runs(project, + status=['RUNNING', 'FAILED', 'TERMINATED', 'CANCELLED'], + heartbeat_before=cutoff_iso) + + For a literal ``running OR recent-heartbeat`` union, issue two calls and + union the run IDs client-side. Returns: List of run dicts with keys: ``id``, ``name``, ``displayId``, @@ -293,13 +317,27 @@ def list_runs( ``url``. Raises: - ValueError: If more than 50 ``field_filters`` terms are given. + ValueError: If more than 50 ``field_filters`` terms are given, or a + ``status`` value is not a recognised run status. """ params: Dict[str, Any] = {'projectName': project, 'limit': min(limit, 200)} if search is not None: params['search'] = search if tags is not None: params['tags'] = ','.join(tags) + if status is not None: + invalid = [s for s in status if s not in _RUN_STATUSES] + if invalid: + raise ValueError( + f'invalid status value(s) {invalid}; ' + f'allowed: {sorted(_RUN_STATUSES)}' + ) + if status: + params['status'] = ','.join(status) + if heartbeat_after is not None: + params['heartbeatAfter'] = heartbeat_after + if heartbeat_before is not None: + params['heartbeatBefore'] = heartbeat_before if field_filters is not None: if len(field_filters) > _MAX_FILTER_TERMS: raise ValueError( @@ -750,6 +788,9 @@ def list_runs( field_filters: Optional[List[Union['FieldFilter', Dict[str, Any]]]] = None, sort: Optional[str] = None, offset: int = 0, + status: Optional[List[str]] = None, + heartbeat_after: Optional[str] = None, + heartbeat_before: Optional[str] = None, ) -> List[Dict[str, Any]]: """List runs in a project. See :meth:`Client.list_runs`.""" return _get_client().list_runs( @@ -760,6 +801,9 @@ def list_runs( field_filters=field_filters, sort=sort, offset=offset, + status=status, + heartbeat_after=heartbeat_after, + heartbeat_before=heartbeat_before, ) diff --git a/tests/test_query.py b/tests/test_query.py index 1c2b7bf..349a1a5 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -237,6 +237,28 @@ def test_negative_offset_omitted(self, client, mock_response): call_args = client._client.get.call_args assert 'offset' not in call_args[1]['params'] + def test_status_filter(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + client.list_runs('proj', status=['RUNNING', 'FAILED']) + call_args = client._client.get.call_args + assert call_args[1]['params']['status'] == 'RUNNING,FAILED' + + def test_bad_status_raises(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + with pytest.raises(ValueError, match='invalid status'): + client.list_runs('proj', status=['RUNNING', 'NOPE']) + + def test_heartbeat_filters(self, client, mock_response): + client._client.get.return_value = mock_response(200, {'runs': []}) + client.list_runs( + 'proj', + heartbeat_after='2026-06-01T00:00:00Z', + heartbeat_before='2026-06-22T00:00:00Z', + ) + params = client._client.get.call_args[1]['params'] + assert params['heartbeatAfter'] == '2026-06-01T00:00:00Z' + assert params['heartbeatBefore'] == '2026-06-22T00:00:00Z' + def test_field_filters_objects(self, client, mock_response): from pluto.query import FieldFilter @@ -680,6 +702,9 @@ def test_list_runs_creates_default_client(self, monkeypatch, mock_response): field_filters=None, sort=None, offset=0, + status=None, + heartbeat_after=None, + heartbeat_before=None, ) # Clean up From ba8bf8dad7f6d247db6cb87acc80d785f7f1aa69 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 23 Jun 2026 23:36:53 +0000 Subject: [PATCH 09/10] query: replace filter kwargs with a wandb-style filters= dict Consolidates list_runs filtering behind a single OR-capable `filters` argument (wandb's api.runs(filters=...) shape) instead of one kwarg per dimension. Removes the field_filters/status/heartbeat_* kwargs and the public FieldFilter dataclass added earlier on this branch (unmerged), in favor of: list_runs(project, filters={"$or": [ {"state": "running"}, {"heartbeat_at": {"$gte": cutoff}}, ]}) - list_runs(project, search, tags, limit, sort, offset, filters): search/tags/ sort/offset kept; filters is a MongoDB-style dict sent as the `filter` param. - _validate_filters: light client-side structural check of boolean/leaf operators + field names (clear ValueError; server enforces full semantics). - test_contract.py reframed: assert the deployed server documents the `filter` param (skips until deployed) instead of the now-internal FieldFilterTerm enum. - Tests + e2e updated to the filters surface; docs regenerated. Companion to server #512 (run-filter compiler). Co-Authored-By: Claude Opus 4.8 (1M context) --- docs-api/query.mdx | 17 +-- pluto/query.py | 308 ++++++++++++++--------------------------- tests/test_contract.py | 98 +++++-------- tests/test_e2e.py | 10 +- tests/test_query.py | 175 ++++++----------------- 5 files changed, 189 insertions(+), 419 deletions(-) diff --git a/docs-api/query.mdx b/docs-api/query.mdx index d8fc424..8f946d4 100644 --- a/docs-api/query.mdx +++ b/docs-api/query.mdx @@ -44,12 +44,9 @@ list_runs( search: Optional[str] = None, tags: Optional[List[str]] = None, limit: int = 50, - field_filters: Optional[List[Union[FieldFilter, Dict[str, Any]]]] = None, sort: Optional[str] = None, offset: int = 0, - status: Optional[List[str]] = None, - heartbeat_after: Optional[str] = None, - heartbeat_before: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]] ``` @@ -61,18 +58,15 @@ List runs in a project. | `search` | `Optional[str]` | Full-text search on run name. | | `tags` | `Optional[List[str]]` | Filter by tags (AND logic). Only runs matching *all* specified tags are returned. | | `limit` | `int` | Maximum number of runs to return (max 200). | -| `field_filters` | `Optional[List[Union[FieldFilter, Dict[str, Any]]]]` | Server-side filters over `config.*` / `systemMetadata.*` fields. A list of `FieldFilter` instances (or raw dicts in the same `{source, key, dataType, operator, values}` shape). Terms are AND-combined. At most 50 terms. | | `sort` | `Optional[str]` | Server-side ordering as `[+\|-]` (`-` = descending). Accepts built-in columns (`createdAt`, `updatedAt`, `name`, `status` and their snake_case aliases), `config..value` / `systemMetadata..value`, `summary_metrics.` (LAST aggregation), and `heartbeat_at` (last logged data point). Defaults to `createdAt desc` server-side. | | `offset` | `int` | Skip-based pagination offset (0–100,000). Combine with `limit` to page; advance `offset` until a short page is returned. | -| `status` | `Optional[List[str]]` | Keep only runs whose status is in this list (OR within the list). Allowed: `RUNNING`, `COMPLETED`, `FAILED`, `TERMINATED`, `CANCELLED`. | -| `heartbeat_after` | `Optional[str]` | ISO-8601 timestamp; keep only runs whose last logged-data time (heartbeat) is at or after this. Requires `project`. | -| `heartbeat_before` | `Optional[str]` | ISO-8601 timestamp; keep only runs whose heartbeat is at or before this (i.e. *stale*). Requires `project`. | +| `filters` | `Optional[Dict[str, Any]]` | A wandb-compatible MongoDB-style query (the OR-capable filter surface). Boolean `$and`/`$or`/`$not` combine leaf terms; each leaf is `{field: value}` (equality) or `{field: {$op: value}}` with ops `$eq`, `$ne`, `$gt`, `$gte`, `$lt`, `$lte`, `$in`, `$nin`, `$regex`. Fields: `state`/`status`, `heartbeat_at`, `created_at`/`updated_at`, `name`, `tags`, `config.`, `summaryMetrics.`. `$or`/`$not` and `heartbeat_at`/`summaryMetrics.*` leaves require `project`. | **Returns:** `List[Dict[str, Any]]` — List of run dicts with keys: `id`, `name`, `displayId`, `status`, `tags`, `config`, `createdAt`, `updatedAt`, `url`. **Raises:** -- `ValueError` — If more than 50 `field_filters` terms are given, or a `status` value is not a recognised run status. +- `ValueError` — If `filters` uses an unrecognised operator or field. #### `Client.get_run` @@ -291,12 +285,9 @@ list_runs( search: Optional[str] = None, tags: Optional[List[str]] = None, limit: int = 50, - field_filters: Optional[List[Union[FieldFilter, Dict[str, Any]]]] = None, sort: Optional[str] = None, offset: int = 0, - status: Optional[List[str]] = None, - heartbeat_after: Optional[str] = None, - heartbeat_before: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]] ``` diff --git a/pluto/query.py b/pluto/query.py index b7d0267..b38f420 100644 --- a/pluto/query.py +++ b/pluto/query.py @@ -23,7 +23,6 @@ import os import time import warnings -from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional, Union @@ -47,148 +46,89 @@ def __init__(self, message: str, status_code: Optional[int] = None): super().__init__(message) -# Field-filter vocabulary, kept in step with the server's zod schema for the -# ``/api/runs/list`` ``fieldFilters`` parameter (pluto-server -# ``buildValueCondition``). Operators are scoped to the dataType they apply to — -# the server dispatches on dataType — so e.g. ``contains`` is text-only and ``>`` -# is number-only. ``exists``/``not exists`` work for any dataType. -_FILTER_SOURCES = {'config', 'systemMetadata'} -_FILTER_DATATYPES = {'text', 'number', 'date', 'option'} -_FILTER_OPERATORS_COMMON = {'exists', 'not exists'} -_FILTER_OPERATORS_BY_DATATYPE = { - 'text': { - 'contains', - 'does not contain', - 'equals', - 'is', - 'is not', - 'starts with', - 'ends with', - 'regex', - }, - 'number': { - 'is', - 'is not', - 'is greater than', - '>', - 'is less than', - '<', - 'is greater than or equal to', - '>=', - 'is less than or equal to', - '<=', - 'is between', - 'is not between', - }, - 'date': { - 'is before', - 'is on or before', - 'is after', - 'is on or after', - 'is between', - 'is not between', - }, - 'option': {'is', 'is not', 'is any of', 'is none of'}, +# wandb-compatible vocabulary for the ``filters`` query (the ``/api/runs/list`` +# ``filter`` param). The server (pluto-server ``lib/queries/run-filter.ts``) is +# the authority on detailed semantics; the client does a light structural check +# so obvious mistakes fail fast with a clear ``ValueError`` instead of an opaque +# HTTP 400. +_FILTER_LEAF_OPS = { + '$eq', + '$ne', + '$gt', + '$gte', + '$lt', + '$lte', + '$in', + '$nin', + '$regex', } -# Flat union — the single operator enum the server publishes in OpenAPI. -# ``tests/test_contract.py`` asserts this equals -# ``components.schemas.FieldFilterTerm.properties.operator.enum`` in the live -# spec, so drift fails CI rather than surfacing as opaque HTTP 400s. -_FILTER_OPERATORS = _FILTER_OPERATORS_COMMON.union( - *_FILTER_OPERATORS_BY_DATATYPE.values() +# Recognised leaf field names (exact) plus dotted-prefix families. +_FILTER_FIELDS = { + 'state', + 'status', + 'heartbeat_at', + 'heartbeatAt', + 'created_at', + 'createdAt', + 'updated_at', + 'updatedAt', + 'name', + 'displayName', + 'display_name', + 'tags', +} +_FILTER_FIELD_PREFIXES = ( + 'config.', + 'systemMetadata.', + 'summaryMetrics.', + 'summary_metrics.', ) - -def _allowed_operators(data_type: str) -> set: - """Operators valid for *data_type* (its dataType-specific set + common).""" - return ( - _FILTER_OPERATORS_BY_DATATYPE.get(data_type, set()) | _FILTER_OPERATORS_COMMON - ) - - -# Server caps the filter set at 50 terms; reject early with a clear error. -_MAX_FILTER_TERMS = 50 # Server clamps offset to this range (MAX_JSON_SORT_OFFSET). _MAX_OFFSET = 100_000 -# Run lifecycle statuses accepted by the server's ``status`` filter. -_RUN_STATUSES = {'RUNNING', 'COMPLETED', 'FAILED', 'TERMINATED', 'CANCELLED'} - -@dataclass -class FieldFilter: - """A single server-side filter term for :meth:`Client.list_runs`. - Filters runs by an indexed ``config.*`` or ``systemMetadata.*`` field. - Multiple terms are AND-combined by the server. +def _validate_filters(node: Any, _depth: int = 0) -> None: + """Light structural validation of a wandb-style ``filters`` dict. - Args: - source: ``"config"`` or ``"systemMetadata"``. - key: Dotted field path (e.g. ``"checkpoint.r2_prefix"``). - dataType: One of ``"text"``, ``"number"``, ``"date"``, ``"option"``. - operator: An operator valid for *dataType* (the server dispatches on - dataType). text: ``contains``, ``does not contain``, ``equals``, - ``is``, ``is not``, ``starts with``, ``ends with``, ``regex``; - number: ``is``, ``is not``, ``is greater than``/``>``, - ``is less than``/``<``, ``is greater than or equal to``/``>=``, - ``is less than or equal to``/``<=``, ``is between``, - ``is not between``; date: ``is before``, ``is on or before``, - ``is after``, ``is on or after``, ``is between``, ``is not between``; - option: ``is``, ``is not``, ``is any of``, ``is none of``. Any - dataType also accepts ``exists`` / ``not exists``. - values: Operand list. A scalar is coerced to a single-element list. - ``is between`` takes two values; ``exists`` / ``not exists`` take - none. - - Example:: - - FieldFilter("config", "lr", "number", ">", [0.001]) - FieldFilter("config", "lr", "number", "is between", [0.1, 0.9]) - FieldFilter("config", "model", "text", "contains", "gpt") - FieldFilter("config", "checkpoint", "text", "exists") - - Raises: - ValueError: If ``source`` or ``dataType`` is unrecognised, or - ``operator`` is not valid for ``dataType``. + Recursively checks boolean operators (``$and``/``$or``/``$not``), leaf + operators, and field names against the supported vocabulary, raising + ``ValueError`` on anything unrecognised. The server enforces full semantics. """ - - source: str - key: str - dataType: str - operator: str - values: List[Any] = field(default_factory=list) - - def __post_init__(self) -> None: - if self.source not in _FILTER_SOURCES: - raise ValueError( - f'FieldFilter source must be one of {sorted(_FILTER_SOURCES)}, ' - f'got {self.source!r}' - ) - if self.dataType not in _FILTER_DATATYPES: - raise ValueError( - f'FieldFilter dataType must be one of {sorted(_FILTER_DATATYPES)}, ' - f'got {self.dataType!r}' - ) - allowed = _allowed_operators(self.dataType) - if self.operator not in allowed: - raise ValueError( - f'FieldFilter operator {self.operator!r} is not valid for ' - f'dataType {self.dataType!r}; expected one of {sorted(allowed)}' - ) - # Accept a bare scalar for convenience (e.g. operator "is"). - if not isinstance(self.values, (list, tuple)): - self.values = [self.values] + if _depth > 50: + raise ValueError('filters nested too deeply') + if not isinstance(node, dict): + raise ValueError(f'filters node must be a dict, got {type(node).__name__}') + for key, value in node.items(): + if key in ('$and', '$or'): + if not isinstance(value, list): + raise ValueError(f'{key} expects a list') + for child in value: + _validate_filters(child, _depth + 1) + elif key == '$not': + _validate_filters(value, _depth + 1) + elif key.startswith('$'): + raise ValueError(f'unknown boolean operator: {key!r}') else: - self.values = list(self.values) - - def to_dict(self) -> Dict[str, Any]: - """Serialize to the server's filter-term JSON shape.""" - return { - 'source': self.source, - 'key': self.key, - 'dataType': self.dataType, - 'operator': self.operator, - 'values': self.values, - } + _validate_filter_field(key) + if isinstance(value, dict) and any(k.startswith('$') for k in value): + for op in value: + if op not in _FILTER_LEAF_OPS: + raise ValueError( + f'unknown leaf operator {op!r} on field {key!r}; ' + f'expected one of {sorted(_FILTER_LEAF_OPS)}' + ) + + +def _validate_filter_field(field_name: str) -> None: + if field_name in _FILTER_FIELDS: + return + if any( + field_name.startswith(p) and len(field_name) > len(p) + for p in _FILTER_FIELD_PREFIXES + ): + return + raise ValueError(f'unknown filter field: {field_name!r}') class Client: @@ -261,12 +201,9 @@ def list_runs( search: Optional[str] = None, tags: Optional[List[str]] = None, limit: int = 50, - field_filters: Optional[List[Union['FieldFilter', Dict[str, Any]]]] = None, sort: Optional[str] = None, offset: int = 0, - status: Optional[List[str]] = None, - heartbeat_after: Optional[str] = None, - heartbeat_before: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]]: """List runs in a project. @@ -276,11 +213,6 @@ def list_runs( tags: Filter by tags (AND logic). Only runs matching *all* specified tags are returned. limit: Maximum number of runs to return (max 200). - field_filters: Server-side filters over ``config.*`` / - ``systemMetadata.*`` fields. A list of :class:`FieldFilter` - instances (or raw dicts in the same ``{source, key, dataType, - operator, values}`` shape). Terms are AND-combined. At most 50 - terms. sort: Server-side ordering as ``[+|-]`` (``-`` = descending). Accepts built-in columns (``createdAt``, ``updatedAt``, ``name``, ``status`` and their snake_case aliases), @@ -291,25 +223,32 @@ def list_runs( offset: Skip-based pagination offset (0–100,000). Combine with ``limit`` to page; advance ``offset`` until a short page is returned. - status: Keep only runs whose status is in this list (OR within the - list). Allowed: ``RUNNING``, ``COMPLETED``, ``FAILED``, - ``TERMINATED``, ``CANCELLED``. - heartbeat_after: ISO-8601 timestamp; keep only runs whose last - logged-data time (heartbeat) is at or after this. Requires - ``project``. - heartbeat_before: ISO-8601 timestamp; keep only runs whose heartbeat - is at or before this (i.e. *stale*). Requires ``project``. - - All filters AND together. To find interrupted jobs to retry (the wandb - "stale spot run" pattern), select non-completed runs that haven't - reported data recently:: - - list_runs(project, - status=['RUNNING', 'FAILED', 'TERMINATED', 'CANCELLED'], - heartbeat_before=cutoff_iso) - - For a literal ``running OR recent-heartbeat`` union, issue two calls and - union the run IDs client-side. + filters: A wandb-compatible MongoDB-style query (the OR-capable + filter surface). Boolean ``$and``/``$or``/``$not`` combine leaf + terms; each leaf is ``{field: value}`` (equality) or + ``{field: {$op: value}}`` with ops ``$eq``, ``$ne``, ``$gt``, + ``$gte``, ``$lt``, ``$lte``, ``$in``, ``$nin``, ``$regex``. + Fields: ``state``/``status``, ``heartbeat_at``, + ``created_at``/``updated_at``, ``name``, ``tags``, + ``config.``, ``summaryMetrics.``. + ``$or``/``$not`` and ``heartbeat_at``/``summaryMetrics.*`` leaves + require ``project``. + + ``filters`` AND-combines with ``search``/``tags``. To find interrupted + spot jobs to retry, find the *alive* set and exclude it — or directly + select non-completed runs that have gone stale:: + + # wandb-style "alive" set (running OR reported data in the last hour): + list_runs(project, filters={"$or": [ + {"state": "running"}, + {"heartbeat_at": {"$gte": cutoff_iso}}, + ]}) + + # retry candidates directly (not completed AND stale): + list_runs(project, filters={"$and": [ + {"status": {"$ne": "COMPLETED"}}, + {"heartbeat_at": {"$lt": cutoff_iso}}, + ]}) Returns: List of run dicts with keys: ``id``, ``name``, ``displayId``, @@ -317,45 +256,16 @@ def list_runs( ``url``. Raises: - ValueError: If more than 50 ``field_filters`` terms are given, or a - ``status`` value is not a recognised run status. + ValueError: If ``filters`` uses an unrecognised operator or field. """ params: Dict[str, Any] = {'projectName': project, 'limit': min(limit, 200)} if search is not None: params['search'] = search if tags is not None: params['tags'] = ','.join(tags) - if status is not None: - invalid = [s for s in status if s not in _RUN_STATUSES] - if invalid: - raise ValueError( - f'invalid status value(s) {invalid}; ' - f'allowed: {sorted(_RUN_STATUSES)}' - ) - if status: - params['status'] = ','.join(status) - if heartbeat_after is not None: - params['heartbeatAfter'] = heartbeat_after - if heartbeat_before is not None: - params['heartbeatBefore'] = heartbeat_before - if field_filters is not None: - if len(field_filters) > _MAX_FILTER_TERMS: - raise ValueError( - f'At most {_MAX_FILTER_TERMS} field_filters terms are ' - f'supported, got {len(field_filters)}' - ) - terms = [] - for f in field_filters: - if isinstance(f, FieldFilter): - terms.append(f.to_dict()) - elif isinstance(f, dict): - terms.append(f) - else: - raise TypeError( - 'each field_filters item must be a FieldFilter or dict, ' - f'got {type(f).__name__}' - ) - params['fieldFilters'] = json.dumps(terms) + if filters is not None: + _validate_filters(filters) + params['filter'] = json.dumps(filters) if sort is not None: params['sort'] = sort if offset: @@ -785,12 +695,9 @@ def list_runs( search: Optional[str] = None, tags: Optional[List[str]] = None, limit: int = 50, - field_filters: Optional[List[Union['FieldFilter', Dict[str, Any]]]] = None, sort: Optional[str] = None, offset: int = 0, - status: Optional[List[str]] = None, - heartbeat_after: Optional[str] = None, - heartbeat_before: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]]: """List runs in a project. See :meth:`Client.list_runs`.""" return _get_client().list_runs( @@ -798,12 +705,9 @@ def list_runs( search=search, tags=tags, limit=limit, - field_filters=field_filters, sort=sort, offset=offset, - status=status, - heartbeat_after=heartbeat_after, - heartbeat_before=heartbeat_before, + filters=filters, ) diff --git a/tests/test_contract.py b/tests/test_contract.py index c5ad75b..105e9b6 100644 --- a/tests/test_contract.py +++ b/tests/test_contract.py @@ -1,20 +1,20 @@ -"""Contract test: keep the client's field-filter vocabulary in step with the -server's zod schema. - -The server's zod schema for the ``/api/runs/list`` ``fieldFilters`` parameter is -the source of truth. Rather than reimplement validation, we assert the client's -hardcoded enums match the schema the server publishes in its OpenAPI document -(served at ``{url_api}/api/openapi.json``), so any drift fails CI loudly instead -of surfacing as opaque HTTP 400s for users. - -This depends on the server exposing the inner filter-term schema as a structured -OpenAPI component (``components.schemas.FieldFilterTerm`` with real ``enum``s). -Until that companion server change lands, the spec carries the operators only as -prose in the parameter description, so this test SKIPS rather than fails. - -Network test: hits the live API spec endpoint (no auth required for the public -spec). Skips cleanly when the spec is unreachable, so offline/hermetic runs are -unaffected — matching the network-dependent style of ``tests/test_e2e.py``. +"""Contract test: the client's ``filters`` surface has a server counterpart. + +The client sends `list_runs(filters=...)` as the `/api/runs/list` ``filter`` +query param. This test fetches the server's published OpenAPI document +(``{url_api}/api/openapi.json``) and asserts that endpoint actually documents a +``filter`` parameter — so a deployed server that dropped/renamed it (or a client +pointed at a server that predates the feature) fails CI loudly instead of +silently no-op-ing the filter. + +Until the server change is deployed to the target host, ``/api/runs/list`` has no +``filter`` param yet, so the test SKIPS rather than fails. The detailed operator +vocabulary is validated client-side (``tests/test_query.py``) and on the server; +a structured ``RunFilter`` schema for a full operator/field contract is a +fast-follow (see plan). + +Network test: hits the live API spec endpoint. Skips cleanly when unreachable, so +offline/hermetic runs are unaffected — matching the style of ``tests/test_e2e.py``. """ import os @@ -22,15 +22,7 @@ import httpx import pytest -from pluto.query import ( - _FILTER_DATATYPES, - _FILTER_OPERATORS, - _FILTER_SOURCES, - _resolve_url_api, -) - -# Component name the companion server PR registers in its OpenAPI document. -_COMPONENT = 'FieldFilterTerm' +from pluto.query import _resolve_url_api def _fetch_openapi() -> dict: @@ -54,48 +46,26 @@ def _fetch_openapi() -> dict: pytest.skip(f'OpenAPI spec at {url} returned non-JSON body: {exc}') -def _term_schema() -> dict: +def _list_runs_params() -> dict: spec = _fetch_openapi() - schemas = spec.get('components', {}).get('schemas', {}) - if _COMPONENT not in schemas: + op = spec.get('paths', {}).get('/api/runs/list', {}).get('get', {}) + params = {p.get('name') for p in op.get('parameters', [])} + if 'filter' not in params: pytest.skip( - f'OpenAPI spec has no components.schemas.{_COMPONENT} yet; ' - 'pending the companion pluto-server change that surfaces the ' - 'field-filter zod schema as a structured component.' + 'GET /api/runs/list has no `filter` param yet; pending the ' + 'pluto-server filter-query change being deployed to this host.' ) - return schemas[_COMPONENT] - - -def _enum_for(term: dict, field: str) -> set: - prop = term.get('properties', {}).get(field, {}) - enum = prop.get('enum') - if enum is None: - pytest.skip(f'{_COMPONENT}.{field} has no enum in the OpenAPI spec') - return set(enum) - - -def test_filter_operators_match_server(): - server = _enum_for(_term_schema(), 'operator') - assert server == _FILTER_OPERATORS, ( - 'operator enum drift between client and server.\n' - f' server-only: {sorted(server - _FILTER_OPERATORS)}\n' - f' client-only: {sorted(_FILTER_OPERATORS - server)}' - ) + return params -def test_filter_sources_match_server(): - server = _enum_for(_term_schema(), 'source') - assert server == _FILTER_SOURCES, ( - 'source enum drift between client and server.\n' - f' server-only: {sorted(server - _FILTER_SOURCES)}\n' - f' client-only: {sorted(_FILTER_SOURCES - server)}' - ) +def test_filter_param_is_published(): + params = _list_runs_params() + assert 'filter' in params -def test_filter_datatypes_match_server(): - server = _enum_for(_term_schema(), 'dataType') - assert server == _FILTER_DATATYPES, ( - 'dataType enum drift between client and server.\n' - f' server-only: {sorted(server - _FILTER_DATATYPES)}\n' - f' client-only: {sorted(_FILTER_DATATYPES - server)}' - ) +def test_list_runs_still_documents_core_params(): + # Guard against an accidental contract regression on the surface the client + # relies on alongside `filter`. + params = _list_runs_params() + for p in ('projectName', 'limit', 'sort', 'offset'): + assert p in params, f'/api/runs/list missing documented param: {p}' diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 64d1842..6cd9048 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -482,8 +482,8 @@ def test_e2e_list_runs_offset_pagination(): assert [r['id'] for r in page2] == [id_a], 'offset=1 should skip to the older run' -def test_e2e_list_runs_field_filter(): - """Verify fieldFilters filters runs by a config value, server-side.""" +def test_e2e_list_runs_filter(): + """Verify the wandb-style `filters` query filters by a config value.""" marker = f'e2e-ff-{int(time.time())}' run = pluto.init( project=TESTING_PROJECT_NAME, @@ -496,9 +496,7 @@ def test_e2e_list_runs_field_filter(): def _query(): runs = pq.list_runs( TESTING_PROJECT_NAME, - field_filters=[ - pq.FieldFilter('config', 'e2e_filter_marker', 'text', 'is', [marker]) - ], + filters={'config.e2e_filter_marker': marker}, limit=200, ) ids = [r['id'] for r in runs] @@ -507,7 +505,7 @@ def _query(): # Field values are indexed asynchronously; poll for eventual consistency. assert _poll( fn=_query, check=lambda found: found - ), f'Run {run_id} not found via fieldFilters on config.e2e_filter_marker' + ), f'Run {run_id} not found via filters on config.e2e_filter_marker' # --------------------------------------------------------------------------- diff --git a/tests/test_query.py b/tests/test_query.py index 349a1a5..81b2a09 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -237,150 +237,60 @@ def test_negative_offset_omitted(self, client, mock_response): call_args = client._client.get.call_args assert 'offset' not in call_args[1]['params'] - def test_status_filter(self, client, mock_response): + def test_filters_serialized_as_json(self, client, mock_response): client._client.get.return_value = mock_response(200, {'runs': []}) - client.list_runs('proj', status=['RUNNING', 'FAILED']) - call_args = client._client.get.call_args - assert call_args[1]['params']['status'] == 'RUNNING,FAILED' - - def test_bad_status_raises(self, client, mock_response): - client._client.get.return_value = mock_response(200, {'runs': []}) - with pytest.raises(ValueError, match='invalid status'): - client.list_runs('proj', status=['RUNNING', 'NOPE']) + flt = { + '$or': [ + {'state': 'running'}, + {'heartbeat_at': {'$gte': '2026-06-22T00:00:00Z'}}, + ] + } + client.list_runs('proj', filters=flt) + params = client._client.get.call_args[1]['params'] + assert json.loads(params['filter']) == flt - def test_heartbeat_filters(self, client, mock_response): + def test_filters_config_leaf(self, client, mock_response): client._client.get.return_value = mock_response(200, {'runs': []}) - client.list_runs( - 'proj', - heartbeat_after='2026-06-01T00:00:00Z', - heartbeat_before='2026-06-22T00:00:00Z', - ) + client.list_runs('proj', filters={'config.lr': {'$gt': 0.001}}) params = client._client.get.call_args[1]['params'] - assert params['heartbeatAfter'] == '2026-06-01T00:00:00Z' - assert params['heartbeatBefore'] == '2026-06-22T00:00:00Z' - - def test_field_filters_objects(self, client, mock_response): - from pluto.query import FieldFilter + assert json.loads(params['filter']) == {'config.lr': {'$gt': 0.001}} + def test_filters_unknown_field_raises(self, client, mock_response): client._client.get.return_value = mock_response(200, {'runs': []}) - client.list_runs( - 'proj', - field_filters=[FieldFilter('config', 'lr', 'number', '>', [0.001])], - ) - call_args = client._client.get.call_args - sent = json.loads(call_args[1]['params']['fieldFilters']) - assert sent == [ - { - 'source': 'config', - 'key': 'lr', - 'dataType': 'number', - 'operator': '>', - 'values': [0.001], - } - ] + with pytest.raises(ValueError, match='unknown filter field'): + client.list_runs('proj', filters={'bogus': 1}) - def test_field_filters_raw_dicts(self, client, mock_response): + def test_filters_unknown_operator_raises(self, client, mock_response): client._client.get.return_value = mock_response(200, {'runs': []}) - term = { - 'source': 'systemMetadata', - 'key': 'gpu', - 'dataType': 'text', - 'operator': 'contains', - 'values': ['a100'], - } - client.list_runs('proj', field_filters=[term]) - call_args = client._client.get.call_args - assert json.loads(call_args[1]['params']['fieldFilters']) == [term] + with pytest.raises(ValueError, match='unknown leaf operator'): + client.list_runs('proj', filters={'config.lr': {'$bogus': 1}}) - def test_field_filters_invalid_item_type_raises(self, client, mock_response): + def test_filters_unknown_boolean_op_raises(self, client, mock_response): client._client.get.return_value = mock_response(200, {'runs': []}) - with pytest.raises(TypeError, match='FieldFilter or dict'): - client.list_runs('proj', field_filters=['not-a-filter']) + with pytest.raises(ValueError, match='unknown boolean operator'): + client.list_runs('proj', filters={'$xor': [{'state': 'running'}]}) - def test_field_filters_too_many_raises(self, client, mock_response): - from pluto.query import FieldFilter - client._client.get.return_value = mock_response(200, {'runs': []}) - terms = [FieldFilter('config', f'k{i}', 'text', 'is', ['x']) for i in range(51)] - with pytest.raises(ValueError, match='At most 50'): - client.list_runs('proj', field_filters=terms) - - -class TestFieldFilter: - def test_scalar_values_coerced_to_list(self): - from pluto.query import FieldFilter - - f = FieldFilter('config', 'model', 'text', 'is', 'gpt') - assert f.values == ['gpt'] - assert f.to_dict()['values'] == ['gpt'] - - def test_bad_source_raises(self): - from pluto.query import FieldFilter - - with pytest.raises(ValueError, match='source'): - FieldFilter('cfg', 'lr', 'number', '>', [1]) - - def test_bad_datatype_raises(self): - from pluto.query import FieldFilter - - with pytest.raises(ValueError, match='dataType'): - FieldFilter('config', 'lr', 'float', '>', [1]) - - def test_bad_operator_raises(self): - from pluto.query import FieldFilter - - with pytest.raises(ValueError, match='operator'): - FieldFilter('config', 'lr', 'number', 'gt', [1]) - - @pytest.mark.parametrize( - 'source,key,dtype,op,vals', - [ - ('config', 'lr', 'number', 'is between', [0.1, 0.9]), - ('config', 'lr', 'number', 'is greater than', [0.1]), - ('config', 'ts', 'date', 'is before', ['2026-01-01']), - ('config', 'ts', 'date', 'is on or after', ['2026-01-01']), - ('config', 'name', 'text', 'starts with', ['gpt']), - ('systemMetadata', 'gpu', 'option', 'is any of', ['a100', 'h100']), - ('config', 'checkpoint', 'text', 'exists', []), - ('config', 'lr', 'number', 'not exists', []), - ], - ) - def test_valid_operators_accepted(self, source, key, dtype, op, vals): - from pluto.query import FieldFilter - - f = FieldFilter(source, key, dtype, op, vals) - assert f.to_dict()['operator'] == op - - @pytest.mark.parametrize( - 'dtype,op', - [ - ('number', 'contains'), # text-only operator on number - ('date', 'regex'), # text-only operator on date - ('number', 'is any of'), # option-only operator on number - ('option', '>'), # number-only operator on option - ('text', 'is between'), # number-only operator on text - ], - ) - def test_operator_rejected_for_wrong_datatype(self, dtype, op): - from pluto.query import FieldFilter - - with pytest.raises(ValueError, match='not valid for dataType'): - FieldFilter('config', 'k', dtype, op, ['x']) - - def test_flat_operator_set_is_union(self): - # The flat set the contract test compares against must be exactly the - # union of the per-dataType sets plus the common operators. - from pluto.query import ( - _FILTER_OPERATORS, - _FILTER_OPERATORS_BY_DATATYPE, - _FILTER_OPERATORS_COMMON, +class TestValidateFilters: + def test_accepts_nested_boolean(self): + from pluto.query import _validate_filters + + _validate_filters( + { + '$and': [ + {'$or': [{'status': 'RUNNING'}, {'status': 'FAILED'}]}, + {'$not': {'config.lr': {'$lt': 0.1}}}, + {'tags': {'$in': ['a', 'b']}}, + {'summaryMetrics.loss': {'$lte': 0.5}}, + ] + } ) - expected = set(_FILTER_OPERATORS_COMMON) - for ops in _FILTER_OPERATORS_BY_DATATYPE.values(): - expected |= ops - assert _FILTER_OPERATORS == expected - assert len(_FILTER_OPERATORS) == 26 + def test_and_or_require_list(self): + from pluto.query import _validate_filters + + with pytest.raises(ValueError, match=r'\$or expects a list'): + _validate_filters({'$or': {'state': 'running'}}) # --------------------------------------------------------------------------- @@ -699,12 +609,9 @@ def test_list_runs_creates_default_client(self, monkeypatch, mock_response): search=None, tags=None, limit=50, - field_filters=None, sort=None, offset=0, - status=None, - heartbeat_after=None, - heartbeat_before=None, + filters=None, ) # Clean up From 57a07f0258599bdcabd6ec4bab936874c371d9c1 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Fri, 26 Jun 2026 22:04:39 +0000 Subject: [PATCH 10/10] query: contract-test filter vocabulary against server RunFilterGrammar The filter grammar is now defined canonically on the server (run-filter-grammar.ts) and published as the RunFilterGrammar OpenAPI component. These client constants are the mirror; assert they match the published grammar so drift fails CI instead of surfacing as a runtime 400. - add _FILTER_BOOL_OPS so all four vocab sets (bool/leaf ops, fields, prefixes) are explicit and contract-testable. - test_contract.py: assert each client set equals RunFilterGrammar's published enums (skips until the server change is deployed to the target host). Companion to server #512 (RunFilterGrammar). Co-Authored-By: Claude Opus 4.8 (1M context) --- pluto/query.py | 11 +++++---- tests/test_contract.py | 51 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 57 insertions(+), 5 deletions(-) diff --git a/pluto/query.py b/pluto/query.py index b38f420..c9a25df 100644 --- a/pluto/query.py +++ b/pluto/query.py @@ -47,10 +47,13 @@ def __init__(self, message: str, status_code: Optional[int] = None): # wandb-compatible vocabulary for the ``filters`` query (the ``/api/runs/list`` -# ``filter`` param). The server (pluto-server ``lib/queries/run-filter.ts``) is -# the authority on detailed semantics; the client does a light structural check -# so obvious mistakes fail fast with a clear ``ValueError`` instead of an opaque -# HTTP 400. +# ``filter`` param). The server defines this grammar canonically in +# ``lib/queries/run-filter-grammar.ts`` and publishes it as the +# ``RunFilterGrammar`` OpenAPI component; these constants are the client mirror, +# kept honest by ``tests/test_contract.py`` (drift fails CI). The server is the +# authority on detailed semantics; the client does a light structural check so +# obvious mistakes fail fast with a clear ``ValueError`` instead of an HTTP 400. +_FILTER_BOOL_OPS = {'$and', '$or', '$not'} _FILTER_LEAF_OPS = { '$eq', '$ne', diff --git a/tests/test_contract.py b/tests/test_contract.py index 105e9b6..f93affc 100644 --- a/tests/test_contract.py +++ b/tests/test_contract.py @@ -22,7 +22,13 @@ import httpx import pytest -from pluto.query import _resolve_url_api +from pluto.query import ( + _FILTER_BOOL_OPS, + _FILTER_FIELD_PREFIXES, + _FILTER_FIELDS, + _FILTER_LEAF_OPS, + _resolve_url_api, +) def _fetch_openapi() -> dict: @@ -69,3 +75,46 @@ def test_list_runs_still_documents_core_params(): params = _list_runs_params() for p in ('projectName', 'limit', 'sort', 'offset'): assert p in params, f'/api/runs/list missing documented param: {p}' + + +def _grammar() -> dict: + """The server's published RunFilterGrammar component (its `properties`).""" + spec = _fetch_openapi() + schemas = spec.get('components', {}).get('schemas', {}) + grammar = schemas.get('RunFilterGrammar') + if grammar is None: + pytest.skip( + 'OpenAPI spec has no RunFilterGrammar component yet; pending the ' + 'pluto-server grammar-publish change being deployed to this host.' + ) + return grammar.get('properties', {}) + + +def _published(props: dict, key: str) -> set: + enum = props.get(key, {}).get('items', {}).get('enum') + if enum is None: + pytest.skip(f'RunFilterGrammar.{key} has no items.enum in the spec') + return set(enum) + + +@pytest.mark.parametrize( + 'prop,client', + [ + ('booleanOperators', _FILTER_BOOL_OPS), + ('leafOperators', _FILTER_LEAF_OPS), + ('fields', _FILTER_FIELDS), + ('fieldPrefixes', set(_FILTER_FIELD_PREFIXES)), + ], +) +def test_client_filter_vocab_matches_server(prop, client): + """The client's filter vocabulary must equal the server's published grammar. + + Drift (a field/operator added on one side only) fails here with the exact + set difference, instead of surfacing as a runtime 400. + """ + server = _published(_grammar(), prop) + assert client == server, ( + f'{prop} drift between client and server.\n' + f' server-only: {sorted(server - client)}\n' + f' client-only: {sorted(client - server)}' + )