diff --git a/src/arc_client/management/async_continuous_query.py b/src/arc_client/management/async_continuous_query.py index fc3e1fb..dda4c26 100644 --- a/src/arc_client/management/async_continuous_query.py +++ b/src/arc_client/management/async_continuous_query.py @@ -53,7 +53,7 @@ async def create( data = response.json() if not data.get("success", True): raise ArcError(data.get("error", "Failed to create CQ")) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except ArcError: raise except Exception as e: @@ -72,7 +72,9 @@ async def list( response = await self._http.get("/api/v1/continuous_queries", params=params) data = response.json() - queries = data.get("queries", []) + # Arc answers this one with a bare array; the envelope form is + # tolerated so a body that grows one keeps working. + queries = data if isinstance(data, list) else data.get("queries", []) return [ContinuousQuery(**q) for q in queries] except Exception as e: raise ArcError(f"Failed to list continuous queries: {e}") from e @@ -87,7 +89,7 @@ async def get(self, query_id: int) -> ContinuousQuery: if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except (ArcNotFoundError, ArcError): raise except Exception as e: @@ -98,28 +100,50 @@ async def update( query_id: int, name: Optional[str] = None, description: Optional[str] = None, + database: Optional[str] = None, + source_measurement: Optional[str] = None, + destination_measurement: Optional[str] = None, query: Optional[str] = None, interval: Optional[str] = None, + tag_columns: Optional[List[str]] = None, retention_days: Optional[int] = None, delete_source_after_days: Optional[int] = None, is_active: Optional[bool] = None, ) -> ContinuousQuery: - """Update a continuous query.""" - payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if description is not None: - payload["description"] = description - if query is not None: - payload["query"] = query - if interval is not None: - payload["interval"] = interval - if retention_days is not None: - payload["retention_days"] = retention_days - if delete_source_after_days is not None: - payload["delete_source_after_days"] = delete_source_after_days - if is_active is not None: - payload["is_active"] = is_active + """Update a continuous query. + + Server side, ``PUT`` replaces the whole definition: every column is + written from the body, so a field the body leaves out is stored blank + or zeroed. Arc requires ``name``, ``database``, ``source_measurement``, + ``destination_measurement``, ``query`` and ``interval`` in the body + (Basekick-Labs/arc#993, #1011), and overwrites ``description``, + ``tag_columns``, ``retention_days``, ``delete_source_after_days`` and + ``is_active`` with whatever it carries -- omitting ``is_active`` stops + the continuous query. + + So this reads the current definition and sends it back with the + arguments given here applied: pass only what you want to change. A + value of ``None`` means "leave as it is", which is why there is no way + to clear ``description`` through this method. + """ + current = await self.get(query_id) + changes: dict[str, Any] = { + "name": name, + "description": description, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "query": query, + "interval": interval, + "tag_columns": tag_columns, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + } + payload: dict[str, Any] = { + field: getattr(current, field) if value is None else value + for field, value in changes.items() + } try: response = await self._http.put(f"/api/v1/continuous_queries/{query_id}", json=payload) @@ -129,7 +153,7 @@ async def update( if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except (ArcNotFoundError, ArcError): raise except Exception as e: diff --git a/src/arc_client/management/continuous_query.py b/src/arc_client/management/continuous_query.py index 6f0196a..57f3717 100644 --- a/src/arc_client/management/continuous_query.py +++ b/src/arc_client/management/continuous_query.py @@ -53,7 +53,7 @@ def create( data = response.json() if not data.get("success", True): raise ArcError(data.get("error", "Failed to create CQ")) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except ArcError: raise except Exception as e: @@ -72,7 +72,9 @@ def list( response = self._http.get("/api/v1/continuous_queries", params=params) data = response.json() - queries = data.get("queries", []) + # Arc answers this one with a bare array; the envelope form is + # tolerated so a body that grows one keeps working. + queries = data if isinstance(data, list) else data.get("queries", []) return [ContinuousQuery(**q) for q in queries] except Exception as e: raise ArcError(f"Failed to list continuous queries: {e}") from e @@ -87,7 +89,7 @@ def get(self, query_id: int) -> ContinuousQuery: if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except (ArcNotFoundError, ArcError): raise except Exception as e: @@ -98,28 +100,50 @@ def update( query_id: int, name: Optional[str] = None, description: Optional[str] = None, + database: Optional[str] = None, + source_measurement: Optional[str] = None, + destination_measurement: Optional[str] = None, query: Optional[str] = None, interval: Optional[str] = None, + tag_columns: Optional[List[str]] = None, retention_days: Optional[int] = None, delete_source_after_days: Optional[int] = None, is_active: Optional[bool] = None, ) -> ContinuousQuery: - """Update a continuous query.""" - payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if description is not None: - payload["description"] = description - if query is not None: - payload["query"] = query - if interval is not None: - payload["interval"] = interval - if retention_days is not None: - payload["retention_days"] = retention_days - if delete_source_after_days is not None: - payload["delete_source_after_days"] = delete_source_after_days - if is_active is not None: - payload["is_active"] = is_active + """Update a continuous query. + + Server side, ``PUT`` replaces the whole definition: every column is + written from the body, so a field the body leaves out is stored blank + or zeroed. Arc requires ``name``, ``database``, ``source_measurement``, + ``destination_measurement``, ``query`` and ``interval`` in the body + (Basekick-Labs/arc#993, #1011), and overwrites ``description``, + ``tag_columns``, ``retention_days``, ``delete_source_after_days`` and + ``is_active`` with whatever it carries -- omitting ``is_active`` stops + the continuous query. + + So this reads the current definition and sends it back with the + arguments given here applied: pass only what you want to change. A + value of ``None`` means "leave as it is", which is why there is no way + to clear ``description`` through this method. + """ + current = self.get(query_id) + changes: dict[str, Any] = { + "name": name, + "description": description, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "query": query, + "interval": interval, + "tag_columns": tag_columns, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + } + payload: dict[str, Any] = { + field: getattr(current, field) if value is None else value + for field, value in changes.items() + } try: response = self._http.put(f"/api/v1/continuous_queries/{query_id}", json=payload) @@ -129,7 +153,7 @@ def update( if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except (ArcNotFoundError, ArcError): raise except Exception as e: diff --git a/src/arc_client/models/continuous_query.py b/src/arc_client/models/continuous_query.py index 04909d2..876bea9 100644 --- a/src/arc_client/models/continuous_query.py +++ b/src/arc_client/models/continuous_query.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Optional +from typing import Any, List, Optional from pydantic import BaseModel @@ -18,6 +18,7 @@ class ContinuousQuery(BaseModel): destination_measurement: str query: str interval: str + tag_columns: Optional[List[str]] = None retention_days: Optional[int] = None delete_source_after_days: Optional[int] = None is_active: bool = True @@ -28,6 +29,21 @@ class ContinuousQuery(BaseModel): created_at: Optional[str] = None updated_at: Optional[str] = None + @classmethod + def from_response(cls, data: dict[str, Any]) -> ContinuousQuery: + """Build a definition from a create, get or update response body. + + Arc returns the definition flat, and the flat body has a ``query`` key + holding SQL -- so unwrapping ``data["query"]`` unconditionally passes a + string where a mapping belongs and every one of those three calls fails + with a TypeError. Only treat ``query`` as the envelope when it actually + carries a mapping. + """ + nested = data.get("query") + if isinstance(nested, dict): + return cls(**nested) + return cls(**data) + class ExecuteCQResponse(BaseModel): """Response from executing a continuous query.""" diff --git a/tests/unit/test_management.py b/tests/unit/test_management.py index 1a6aafa..9965e07 100644 --- a/tests/unit/test_management.py +++ b/tests/unit/test_management.py @@ -399,3 +399,143 @@ async def test_empty_where_raises_error(self) -> None: with pytest.raises(ArcValidationError, match="where clause is required"): await client.delete("default", "cpu", "") + + +class _StubResponse: + """Minimal stand-in for the HTTP response the clients unwrap.""" + + def __init__(self, payload: dict) -> None: + self._payload = payload + + def json(self) -> dict: + return self._payload + + +class _StubHTTP: + """Records the PUT body so the update contract can be asserted.""" + + def __init__(self, stored: dict, listing: object | None = None) -> None: + self.stored = stored + self.listing = [stored] if listing is None else listing + self.put_body: dict | None = None + + def get(self, path: str, params: dict | None = None) -> _StubResponse: + if path.endswith("/continuous_queries"): + return _StubResponse(self.listing) + return _StubResponse(self.stored) + + def put(self, path: str, json: dict) -> _StubResponse: + self.put_body = json + return _StubResponse({**self.stored, **json}) + + +class TestContinuousQueryUpdate: + """PUT replaces the whole definition server side, so update must send one. + + Arc requires name, database, source_measurement, destination_measurement, + query and interval in the body (Basekick-Labs/arc#993, #1011) and + overwrites the rest with whatever it carries, so a sparse body either 400s + or silently blanks stored columns. + """ + + STORED = { + "id": 7, + "name": "downsample-cpu", + "description": "hourly rollup", + "database": "default", + "source_measurement": "cpu", + "destination_measurement": "cpu_1h", + "query": ( + "SELECT avg(usage) FROM default.cpu WHERE time >= {start_time} AND time < {end_time}" + ), + "interval": "1h", + "tag_columns": ["host"], + "retention_days": 365, + "delete_source_after_days": 7, + "is_active": True, + } + + def _client(self) -> tuple: + from arc_client.management.continuous_query import ContinuousQueryClient + + http = _StubHTTP(dict(self.STORED)) + return ContinuousQueryClient(http, None), http # type: ignore[arg-type] + + def test_update_sends_every_required_field(self) -> None: + """A one-field change still carries the whole definition.""" + client, http = self._client() + client.update(7, is_active=False) + + assert http.put_body is not None + for field in ( + "name", + "database", + "source_measurement", + "destination_measurement", + "query", + "interval", + ): + assert http.put_body[field] == self.STORED[field], field + + def test_update_preserves_the_fields_arc_overwrites_unconditionally(self) -> None: + """Omitting these would zero them, and is_active would stop the CQ.""" + client, http = self._client() + client.update(7, interval="30m") + + assert http.put_body is not None + assert http.put_body["interval"] == "30m" + assert http.put_body["description"] == "hourly rollup" + assert http.put_body["tag_columns"] == ["host"] + assert http.put_body["retention_days"] == 365 + assert http.put_body["delete_source_after_days"] == 7 + assert http.put_body["is_active"] is True + + def test_update_applies_only_what_it_is_given(self) -> None: + """Every other field keeps its stored value.""" + client, http = self._client() + client.update(7, destination_measurement="cpu_5m", tag_columns=["host", "region"]) + + assert http.put_body is not None + assert http.put_body["destination_measurement"] == "cpu_5m" + assert http.put_body["tag_columns"] == ["host", "region"] + assert http.put_body["source_measurement"] == "cpu" + assert http.put_body["name"] == "downsample-cpu" + + def test_from_response_reads_the_flat_body_arc_returns(self) -> None: + """create, get and update all unwrap the same body. + + Arc answers with the definition flat, and a flat body carries a + ``query`` key holding SQL -- so unwrapping ``data["query"]`` + unconditionally passed a string where a mapping belongs and raised + TypeError on every one of those three calls. + """ + cq = ContinuousQuery.from_response(dict(self.STORED)) + + assert cq.id == 7 + assert cq.query.startswith("SELECT avg(usage)") + assert cq.tag_columns == ["host"] + + def test_from_response_still_accepts_a_nested_body(self) -> None: + """A body that really does nest the definition keeps working.""" + cq = ContinuousQuery.from_response({"success": True, "query": dict(self.STORED)}) + + assert cq.id == 7 + assert cq.destination_measurement == "cpu_1h" + + def test_list_reads_the_bare_array_arc_returns(self) -> None: + """GET /api/v1/continuous_queries answers with an array, not an envelope.""" + client, _ = self._client() + + queries = client.list() + + assert [q.id for q in queries] == [7] + assert queries[0].destination_measurement == "cpu_1h" + + def test_list_still_accepts_an_envelope(self) -> None: + """A body that wraps the array in "queries" keeps working.""" + from arc_client.management.continuous_query import ContinuousQueryClient + + http = _StubHTTP(dict(self.STORED), listing={"queries": [dict(self.STORED)]}) + client = ContinuousQueryClient(http, None) # type: ignore[arg-type] + + assert [q.id for q in client.list()] == [7] diff --git a/uv.lock b/uv.lock index fed2efb..3ff15fc 100644 --- a/uv.lock +++ b/uv.lock @@ -58,8 +58,8 @@ wheels = [ ] [[package]] -name = "arc-client" -version = "0.1.0" +name = "arc-tsdb-client" +version = "0.1.2" source = { editable = "." } dependencies = [ { name = "httpx" },