Add search attribute client parity

Author: durable-workflow-ops

src/durable_workflow/__init__.py

@@ -26,6 +26,7 @@
     ScheduleList,
     ScheduleSpec,
     ScheduleTriggerResult,
+    SearchAttributeList,
     TaskQueueAdmission,
     TaskQueueDescription,
     TaskQueueList,
@@ -163,6 +164,7 @@
     "ScheduleNotFound",
     "ScheduleSpec",
     "ScheduleTriggerResult",
+    "SearchAttributeList",
     "StartChildWorkflow",
     "TaskQueueAdmission",
     "TaskQueueDescription",

src/durable_workflow/client.py

@@ -70,6 +70,8 @@ def _route_for_metrics(path: str) -> str:
             parts[3] = "{run_id}"
     elif parts[0] == "schedules" and len(parts) >= 2:
         parts[1] = "{schedule_id}"
+    elif parts[0] == "search-attributes" and len(parts) >= 2:
+        parts[1] = "{name}"
     elif parts[:2] == ["bridge-adapters", "webhook"] and len(parts) >= 3:
         parts[2] = "{adapter}"
     elif (
@@ -206,6 +208,24 @@ class WorkflowRunList:
     runs: list[WorkflowRun]
 
 
+@dataclass
+class SearchAttributeList:
+    """Search attribute definitions available in the current namespace."""
+
+    system_attributes: dict[str, str]
+    custom_attributes: dict[str, str]
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> SearchAttributeList:
+        system = data.get("system_attributes")
+        custom = data.get("custom_attributes")
+
+        return cls(
+            system_attributes=dict(system) if isinstance(system, dict) else {},
+            custom_attributes=dict(custom) if isinstance(custom, dict) else {},
+        )
+
+
 @dataclass
 class TaskQueueTaskAdmission:
     """Workflow/activity admission state for one task queue."""
@@ -998,6 +1018,51 @@ async def describe_task_queue(self, name: str) -> TaskQueueDescription:
             )
         return TaskQueueDescription.from_dict(data)
 
+    # ── Search attributes ─────────────────────────────────────────────
+    async def list_search_attributes(self) -> SearchAttributeList:
+        """List system and custom search attribute definitions for this namespace."""
+        data = await self._request("GET", "/search-attributes")
+        if not isinstance(data, dict):
+            raise ServerError(
+                200,
+                {
+                    "reason": "invalid_search_attribute_response",
+                    "message": f"expected JSON object, got {type(data).__name__}",
+                },
+            )
+        return SearchAttributeList.from_dict(data)
+
+    async def create_search_attribute(self, name: str, attribute_type: str) -> dict[str, Any]:
+        """Register a custom search attribute and return the server response."""
+        data = await self._request(
+            "POST",
+            "/search-attributes",
+            json={"name": name, "type": attribute_type},
+            context=name,
+        )
+        if not isinstance(data, dict):
+            raise ServerError(
+                200,
+                {
+                    "reason": "invalid_search_attribute_response",
+                    "message": f"expected JSON object, got {type(data).__name__}",
+                },
+            )
+        return data
+
+    async def delete_search_attribute(self, name: str) -> dict[str, Any]:
+        """Remove a custom search attribute and return the server response."""
+        data = await self._request("DELETE", f"/search-attributes/{quote(name, safe='')}", context=name)
+        if not isinstance(data, dict):
+            raise ServerError(
+                200,
+                {
+                    "reason": "invalid_search_attribute_response",
+                    "message": f"expected JSON object, got {type(data).__name__}",
+                },
+            )
+        return data
+
     # ── Workflows ──────────────────────────────────────────────────────
     async def start_workflow(
         self,

tests/fixtures/control-plane/search-attribute-create-parity.json

@@ -0,0 +1,36 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "search_attribute.create",
+  "request": {
+    "method": "POST",
+    "path": "/search-attributes",
+    "body": {
+      "name": "OrderStatus",
+      "type": "keyword"
+    }
+  },
+  "semantic_body": {
+    "name": "OrderStatus",
+    "type": "keyword"
+  },
+  "response_body": {
+    "name": "OrderStatus",
+    "type": "keyword",
+    "outcome": "created"
+  },
+  "cli": {
+    "argv": {
+      "name": "OrderStatus",
+      "type": "keyword",
+      "--json": true
+    }
+  },
+  "sdk_python": {
+    "method": "create_search_attribute",
+    "args": {
+      "name": "OrderStatus",
+      "attribute_type": "keyword"
+    }
+  }
+}

tests/fixtures/control-plane/search-attribute-delete-parity.json

@@ -0,0 +1,29 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "search_attribute.delete",
+  "request": {
+    "method": "DELETE",
+    "path": "/search-attributes/OrderStatus"
+  },
+  "semantic_body": {
+    "name": "OrderStatus",
+    "outcome": "deleted"
+  },
+  "response_body": {
+    "name": "OrderStatus",
+    "outcome": "deleted"
+  },
+  "cli": {
+    "argv": {
+      "name": "OrderStatus",
+      "--json": true
+    }
+  },
+  "sdk_python": {
+    "method": "delete_search_attribute",
+    "args": {
+      "name": "OrderStatus"
+    }
+  }
+}

tests/fixtures/control-plane/search-attribute-list-parity.json

@@ -0,0 +1,38 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "search_attribute.list",
+  "request": {
+    "method": "GET",
+    "path": "/search-attributes"
+  },
+  "semantic_body": {
+    "system_attributes": {
+      "WorkflowType": "keyword",
+      "StartTime": "datetime"
+    },
+    "custom_attributes": {
+      "OrderStatus": "keyword",
+      "Priority": "int"
+    }
+  },
+  "response_body": {
+    "system_attributes": {
+      "WorkflowType": "keyword",
+      "StartTime": "datetime"
+    },
+    "custom_attributes": {
+      "OrderStatus": "keyword",
+      "Priority": "int"
+    }
+  },
+  "cli": {
+    "argv": {
+      "--json": true
+    }
+  },
+  "sdk_python": {
+    "method": "list_search_attributes",
+    "args": {}
+  }
+}

tests/test_client.py

@@ -1074,6 +1074,55 @@ async def test_describe_task_queue_rejects_non_object_response(self, client: Cli
             await client.describe_task_queue("orders")
 
 
+class TestSearchAttributes:
+    @pytest.mark.asyncio
+    async def test_list_search_attributes_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "search-attribute-list-parity.json"
+        fixture = json.loads(fixture_path.read_text())
+        resp = _mock_response(200, fixture["response_body"])
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = await client.list_search_attributes(**fixture["sdk_python"]["args"])
+
+        assert mock.call_args.args[:2] == (fixture["request"]["method"], f"/api{fixture['request']['path']}")
+        assert result.system_attributes == fixture["semantic_body"]["system_attributes"]
+        assert result.custom_attributes == fixture["semantic_body"]["custom_attributes"]
+
+    @pytest.mark.asyncio
+    async def test_create_search_attribute_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "search-attribute-create-parity.json"
+        fixture = json.loads(fixture_path.read_text())
+        resp = _mock_response(200, fixture["response_body"])
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = await client.create_search_attribute(**fixture["sdk_python"]["args"])
+
+        assert mock.call_args.args[:2] == (fixture["request"]["method"], f"/api{fixture['request']['path']}")
+        assert mock.call_args.kwargs["json"] == fixture["request"]["body"]
+        assert result == fixture["response_body"]
+
+    @pytest.mark.asyncio
+    async def test_delete_search_attribute_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "search-attribute-delete-parity.json"
+        fixture = json.loads(fixture_path.read_text())
+        resp = _mock_response(200, fixture["response_body"])
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = await client.delete_search_attribute(**fixture["sdk_python"]["args"])
+
+        assert mock.call_args.args[:2] == (fixture["request"]["method"], f"/api{fixture['request']['path']}")
+        assert result == fixture["response_body"]
+
+    @pytest.mark.asyncio
+    async def test_search_attribute_name_is_url_encoded(self, client: Client) -> None:
+        resp = _mock_response(200, {"name": "Customer Status", "outcome": "deleted"})
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            await client.delete_search_attribute("Customer Status")
+
+        assert mock.call_args.args[:2] == ("DELETE", "/api/search-attributes/Customer%20Status")
+
+
 class TestSchedules:
     @pytest.mark.asyncio
     async def test_create_schedule_matches_polyglot_fixture(self, client: Client) -> None: