support for device custom property intervals

foxglove/client/api.py

@@ -74,6 +74,17 @@ def bool_query_param(val: bool) -> Optional[str]:
     return str(val).lower() if val is not None else None
 
 
+def comma_separated_query_param(val: Optional[Union[str, List[str]]]) -> Optional[str]:
+    """
+    Serialize a string or list of strings to a comma-separated query parameter.
+    """
+    if val is None:
+        return None
+    if isinstance(val, list):
+        return ",".join(val) if val else None
+    return val
+
+
 def without_nulls(params: Dict[str, Union[T, None]]) -> Dict[str, T]:
     """
     Filter out `None` values from params
@@ -651,12 +662,9 @@ def get_device(
         :param project_id: Project to retrieve the device from.
             Required for multi-project organizations.
         """
-        if device_name and device_id:
-            raise RuntimeError("device_id and device_name are mutually exclusive")
-        if device_name is None and device_id is None:
-            raise RuntimeError("device_id or device_name must be provided")
+        identifier = _device_identifier(device_id, device_name)
         response = self.__session.get(
-            self.__url__(f"/v1/devices/{device_name or device_id}"),
+            self.__url__(f"/v1/devices/{urlquote(identifier, safe='')}"),
             params={"projectId": project_id} if project_id is not None else None,
         )
 
@@ -728,13 +736,10 @@ def update_device(
         :param project_id: Project to retrieve the device from.
             Required for multi-project organizations.
         """
-        if device_name and device_id:
-            raise RuntimeError("device_id and device_name are mutually exclusive")
-        if device_name is None and device_id is None:
-            raise RuntimeError("device_id or device_name must be provided")
+        identifier = _device_identifier(device_id, device_name)
 
         response = self.__session.patch(
-            self.__url__(f"/v1/devices/{device_name or device_id}"),
+            self.__url__(f"/v1/devices/{urlquote(identifier, safe='')}"),
             params={"projectId": project_id} if project_id is not None else None,
             json=without_nulls({"name": new_name, "properties": properties}),
         )
@@ -758,12 +763,9 @@ def delete_device(
         :param project_id: Project to delete the device from.
             Required for multi-project organizations.
         """
-        if device_name and device_id:
-            raise RuntimeError("device_id and device_name are mutually exclusive")
-        if device_name is None and device_id is None:
-            raise RuntimeError("device_id or device_name must be provided")
+        identifier = _device_identifier(device_id, device_name)
         response = self.__session.delete(
-            self.__url__(f"/v1/devices/{device_name or device_id}"),
+            self.__url__(f"/v1/devices/{urlquote(identifier, safe='')}"),
             params={"projectId": project_id} if project_id is not None else None,
         )
         json_or_raise(response)
@@ -1333,6 +1335,132 @@ def delete_session(
 
         return json_or_raise(response)
 
+    def get_device_custom_property_time_interval(
+        self,
+        *,
+        device_id: Optional[str] = None,
+        device_name: Optional[str] = None,
+        project_id: Optional[str] = None,
+        id: str,
+    ):
+        """
+        Fetches a single device custom property time interval record.
+
+        device_id: The ID of the device to retrieve the time interval record from.
+            Use this or device_name.
+        device_name: The name of the device to retrieve the time interval record from.
+            Use this or device_id.
+        project_id: Project associated with the device. Required for multi-project organizations.
+        id: The ID of the time interval record to fetch.
+        """
+        identifier = _device_identifier(device_id, device_name)
+
+        response = self.__session.get(
+            self.__url__(
+                f"/v1/devices/{urlquote(identifier, safe='')}/property-time-intervals/"
+                f"{urlquote(id, safe='')}"
+            ),
+            params=without_nulls({"projectId": project_id}),
+        )
+        return _device_custom_property_time_interval_dict(json_or_raise(response))
+
+    def get_device_custom_property_time_intervals(
+        self,
+        *,
+        device_id: Optional[str] = None,
+        device_name: Optional[str] = None,
+        project_id: Optional[str] = None,
+        key: Optional[Union[str, List[str]]] = None,
+        start: Optional[datetime.datetime] = None,
+        end: Optional[datetime.datetime] = None,
+        limit: Optional[int] = None,
+        offset: Optional[int] = None,
+    ):
+        """
+        Lists device custom property time intervals.
+
+        device_id: The ID of the device to retrieve time intervals from.
+            Use this or device_name.
+        device_name: The name of the device to retrieve time intervals from.
+            Use this or device_id.
+        project_id: Project associated with the device. Required for multi-project organizations.
+        key: Optional property key or keys to filter by.
+        start: Optionally include intervals active at or after this time.
+        end: Optionally include intervals active before this time.
+        limit: Optionally limit the number of time intervals returned.
+        offset: Optionally offset the time intervals by this many intervals.
+        """
+        identifier = _device_identifier(device_id, device_name)
+
+        params = {
+            "projectId": project_id,
+            "key": comma_separated_query_param(key),
+            "start": start.astimezone().isoformat() if start else None,
+            "end": end.astimezone().isoformat() if end else None,
+            "limit": limit,
+            "offset": offset,
+        }
+
+        response = self.__session.get(
+            self.__url__(
+                f"/v1/devices/{urlquote(identifier, safe='')}/property-time-intervals"
+            ),
+            params={k: v for k, v in params.items() if v is not None},
+        )
+        return [
+            _device_custom_property_time_interval_dict(r)
+            for r in json_or_raise(response)
+        ]
+
+    def update_device_custom_property_time_interval(
+        self,
+        *,
+        device_id: Optional[str] = None,
+        device_name: Optional[str] = None,
+        project_id: Optional[str] = None,
+        key: str,
+        value: Optional[Union[str, bool, float, int, List[str]]] = None,
+        start: datetime.datetime,
+        end: datetime.datetime,
+    ):
+        """
+        Updates device custom property time intervals over a time range.
+
+        The request is treated as an assertion of truth for the given range,
+        so existing intervals may be split, trimmed, or deleted as needed.
+
+        device_id: The ID of the device to update time intervals for.
+            Use this or device_name.
+        device_name: The name of the device to update time intervals for.
+            Use this or device_id.
+        project_id: Project associated with the device. Required for multi-project organizations.
+        key: The property key to update.
+        value: The value to apply over the given time range.
+            When omitted, the value will be treated as explicitly unset for that range.
+        start: Inclusive start of the property's effective time range.
+        end: Exclusive end of the property's effective time range.
+        """
+        identifier = _device_identifier(device_id, device_name)
+
+        params: Dict[str, Any] = {
+            "projectId": project_id,
+            "key": key,
+            "value": value,
+            "start": start.astimezone().isoformat(),
+            "end": end.astimezone().isoformat(),
+        }
+
+        response = self.__session.post(
+            self.__url__(
+                f"/v1/actions/devices/{urlquote(identifier, safe='')}/update-property-time-interval"
+            ),
+            json=without_nulls(params),
+        )
+        # This endpoint returns 204 No Content on success (no body)
+        if response.status_code == 204:
+            return None
+        return json_or_raise(response)
+
 
 def _session_identifier(session_id: Optional[str], session_key: Optional[str]) -> str:
     if session_id is not None and session_key is not None:
@@ -1345,6 +1473,17 @@ def _session_identifier(session_id: Optional[str], session_key: Optional[str]) -
     return identifier
 
 
+def _device_identifier(device_id: Optional[str], device_name: Optional[str]) -> str:
+    if device_id is not None and device_name is not None:
+        raise RuntimeError("device_id and device_name are mutually exclusive")
+    if device_id is None and device_name is None:
+        raise RuntimeError("device_id or device_name must be provided")
+
+    identifier = device_id if device_id is not None else device_name
+    assert identifier is not None, "one of device_id or device_name must be provided"
+    return identifier
+
+
 def _event_dict(json_event):
     return {
         "id": json_event["id"],
@@ -1386,4 +1525,16 @@ def _session_dict(session):
     }
 
 
+def _device_custom_property_time_interval_dict(interval):
+    end = interval.get("end")
+    return {
+        "id": interval["id"],
+        "device_id": interval["deviceId"],
+        "key": interval["key"],
+        "value": interval["value"],
+        "start": arrow.get(interval["start"]).datetime,
+        "end": arrow.get(end).datetime if end else None,
+    }
+
+
 __all__ = ["Client", "CompressionFormat", "FoxgloveException", "OutputFormat"]

tests/test_device_custom_property_time_interval.py

@@ -0,0 +1,160 @@
+import json
+from datetime import datetime
+from urllib.parse import parse_qs, quote, urlparse
+
+import pytest
+import responses
+from faker import Faker
+
+from foxglove.client import Client
+
+from .api_url import api_url
+
+fake = Faker()
+
+
+@responses.activate
+def test_get_device_custom_property_time_interval_quotes_path_and_passes_project_id():
+    device_name = "Device / Name"
+    time_interval_id = "dcph/id"
+    now = datetime.now()
+    responses.add(
+        responses.GET,
+        api_url(
+            f"/v1/devices/{quote(device_name, safe='')}/property-time-intervals/"
+            f"{quote(time_interval_id, safe='')}"
+        ),
+        json={
+            "id": time_interval_id,
+            "deviceId": fake.uuid4(),
+            "key": "env",
+            "value": "prod",
+            "start": now.isoformat(),
+            "end": now.isoformat(),
+        },
+    )
+
+    client = Client("test")
+    response = client.get_device_custom_property_time_interval(
+        device_name=device_name,
+        project_id="project-id",
+        id=time_interval_id,
+    )
+
+    assert response["id"] == time_interval_id
+    assert parse_qs(urlparse(responses.calls[0].request.url).query) == {
+        "projectId": ["project-id"]
+    }
+
+
+def test_get_device_custom_property_time_interval_rejects_multiple_device_selectors():
+    client = Client("test")
+
+    with pytest.raises(RuntimeError) as exception:
+        client.get_device_custom_property_time_interval(
+            device_id="device-id",
+            device_name="device-name",
+            id="time-interval-id",
+        )
+
+    assert str(exception.value) == "device_id and device_name are mutually exclusive"
+
+
+@responses.activate
+def test_get_device_custom_property_time_intervals_uses_path_selector_only():
+    device_name = "Device / Name"
+    responses.add(
+        responses.GET,
+        api_url(f"/v1/devices/{quote(device_name, safe='')}/property-time-intervals"),
+        json=[],
+    )
+
+    client = Client("test")
+    client.get_device_custom_property_time_intervals(
+        device_name=device_name,
+        project_id="project-id",
+        key="env",
+        limit=5,
+        offset=10,
+    )
+
+    assert parse_qs(urlparse(responses.calls[0].request.url).query) == {
+        "key": ["env"],
+        "limit": ["5"],
+        "offset": ["10"],
+        "projectId": ["project-id"],
+    }
+
+
+@responses.activate
+def test_get_device_custom_property_time_intervals_supports_multiple_keys():
+    responses.add(
+        responses.GET,
+        api_url("/v1/devices/device-id/property-time-intervals"),
+        json=[],
+    )
+
+    client = Client("test")
+    client.get_device_custom_property_time_intervals(
+        device_id="device-id",
+        project_id="project-id",
+        key=["env", "region"],
+    )
+
+    assert parse_qs(urlparse(responses.calls[0].request.url).query) == {
+        "key": ["env,region"],
+        "projectId": ["project-id"],
+    }
+
+
+@responses.activate
+def test_update_device_custom_property_time_interval_omits_value_for_clear():
+    device_name = "Device / Name"
+    responses.add(
+        responses.POST,
+        api_url(
+            f"/v1/actions/devices/{quote(device_name, safe='')}/update-property-time-interval"
+        ),
+        status=204,
+        body="",
+    )
+
+    client = Client("test")
+    client.update_device_custom_property_time_interval(
+        device_name=device_name,
+        project_id="project-id",
+        key="env",
+        start=datetime.now(),
+        end=datetime.now(),
+    )
+
+    request_body = json.loads(responses.calls[0].request.body)
+    assert request_body["projectId"] == "project-id"
+    assert request_body["key"] == "env"
+    assert "value" not in request_body
+    assert "deviceName" not in request_body
+    assert "deviceId" not in request_body
+
+
+@responses.activate
+def test_update_device_custom_property_time_interval_supports_array_values():
+    responses.add(
+        responses.POST,
+        api_url("/v1/actions/devices/device-id/update-property-time-interval"),
+        status=204,
+        body="",
+    )
+
+    client = Client("test")
+    client.update_device_custom_property_time_interval(
+        device_id="device-id",
+        key="labels",
+        value=["one", "two"],
+        start=datetime.now(),
+        end=datetime.now(),
+    )
+
+    request_body = json.loads(responses.calls[0].request.body)
+    assert request_body["value"] == ["one", "two"]
+    assert "deviceName" not in request_body
+    assert "deviceId" not in request_body