fix(cq): send the whole definition on update, and read the shapes Arc returns

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:

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:

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."""

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]