[publish] feat: add support for polling/status API

feat: add support for polling/status API [publish]

Author: sadsfae

.readthedocs.yml

@@ -1,4 +1,4 @@
-# See https://docs.readthedocs.io/en/stable/config-file/v2.html for details
+# See https://docs.readthedocs.io/en/latest/config-file/v2.html for details
 version: 2
 sphinx:
   configuration: docs/conf.py

CHANGELOG.rst

@@ -2,6 +2,12 @@
 Changelog
 =========
 
+0.1.16 (2026-06-02)
+-------------------
+
+* Add move status API methods: ``get_all_move_status``, ``get_move_status``,
+  ``start_move_batch``, ``update_move_status``.
+
 0.0.0 (2025-01-07)
 ------------------
 

docs/usage.rst

@@ -36,3 +36,18 @@ You can control certificate verification using the ``verify`` parameter:
 
     # Use a custom CA bundle file
     quads = QuadsApi(username, password, base_url, verify="/path/to/ca-bundle.pem")
+
+Tracking Move Status
+--------------------
+
+Query the 12-stage provisioning pipeline for active host moves:
+
+.. code-block:: python
+
+    with QuadsApi(username, password, base_url) as quads:
+        # All active moves, optionally filtered by cloud or status
+        moves = quads.get_all_move_status(cloud="cloud02")
+
+        # Single host status
+        progress = quads.get_move_status("host01.example.com")
+        print(f"{progress['host']}: {progress['status']}")

setup.py

@@ -13,7 +13,7 @@ def read(*names, **kwargs):
 
 setup(
     name="quads-lib",
-    version="0.1.15",
+    version="0.1.16",
     license="LGPL-3.0-only",
     description="Python client library for interacting with the QUADS API",
     long_description="{}\n{}".format(

src/quads_lib/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.1.15"
+__version__ = "0.1.16"
 
 from .quads import QuadsApi
 

src/quads_lib/quads.py

@@ -367,6 +367,51 @@ def get_moves(self, date: Optional[str] = None) -> dict:
         json_response = self.get(url)
         return json_response
 
+    # Move Status
+    def get_all_move_status(self, cloud: Optional[str] = None, status: Optional[str] = None) -> dict:
+        """Retrieve all active move status records.
+
+        Args:
+            cloud: Filter by target cloud name.
+            status: Filter by move status (e.g. provisioning, failed).
+        """
+        url = "moves/progress/"
+        params = {}
+        if cloud:
+            params["cloud"] = cloud
+        if status:
+            params["status"] = status
+        if params:
+            url = f"{url}?{urlencode(params)}"
+        return self.get(url)
+
+    def get_move_status(self, hostname: str) -> dict:
+        """Retrieve move status for a specific host.
+
+        Args:
+            hostname: The host to query status for.
+        """
+        endpoint = Path("moves") / "progress" / hostname
+        return self.get(str(endpoint))
+
+    def start_move_batch(self, hostnames: list) -> dict:
+        """Start move tracking for a batch of hosts. Requires admin auth.
+
+        Args:
+            hostnames: List of hostnames to start tracking.
+        """
+        return self.post("moves/progress/batch", {"hostnames": hostnames})
+
+    def update_move_status(self, schedule_id: int, data: dict) -> dict:
+        """Update move status on a schedule. Requires admin auth.
+
+        Args:
+            schedule_id: The schedule ID to update.
+            data: Dict with any of status, message, error_message.
+        """
+        endpoint = Path("moves") / "progress" / str(schedule_id)
+        return self.patch(str(endpoint), data)
+
     def get_version(self) -> dict:
         json_response = self.get("version")
         return json_response

tests/test_quads.py

@@ -1763,6 +1763,100 @@ def test_get_moves_with_date(self, mock_get):
         assert str(mock_get.call_args[0][1]).endswith(f"/moves?date={date}")
         assert result == expected_response
 
+    # Move Status
+    @patch("requests.Session.request")
+    def test_get_all_move_status(self, mock_get):
+        expected_response = [
+            {"id": 1, "host": "host1", "status": "pending"},
+            {"id": 2, "host": "host2", "status": "ipmi_config"},
+        ]
+        mock_response = Mock()
+        mock_response.json.return_value = expected_response
+        mock_get.return_value = mock_response
+
+        result = self.api.get_all_move_status()
+
+        mock_get.assert_called_once()
+        assert str(mock_get.call_args[0][1]).endswith("/moves/progress/")
+        assert result == expected_response
+
+    @patch("requests.Session.request")
+    def test_get_all_move_status_with_cloud(self, mock_get):
+        expected_response = [{"id": 1, "host": "host1", "status": "pending"}]
+        mock_response = Mock()
+        mock_response.json.return_value = expected_response
+        mock_get.return_value = mock_response
+
+        result = self.api.get_all_move_status(cloud="cloud02")
+
+        mock_get.assert_called_once()
+        assert "cloud=cloud02" in str(mock_get.call_args[0][1])
+        assert result == expected_response
+
+    @patch("requests.Session.request")
+    def test_get_all_move_status_with_status(self, mock_get):
+        expected_response = [{"id": 1, "host": "host1", "status": "provisioning"}]
+        mock_response = Mock()
+        mock_response.json.return_value = expected_response
+        mock_get.return_value = mock_response
+
+        result = self.api.get_all_move_status(status="provisioning")
+
+        mock_get.assert_called_once()
+        assert "status=provisioning" in str(mock_get.call_args[0][1])
+        assert result == expected_response
+
+    @patch("requests.Session.request")
+    def test_get_move_status(self, mock_get):
+        expected_response = {"id": 1, "host": "host1", "status": "provisioning"}
+        mock_response = Mock()
+        mock_response.json.return_value = expected_response
+        mock_get.return_value = mock_response
+
+        result = self.api.get_move_status("host1")
+
+        mock_get.assert_called_once()
+        assert str(mock_get.call_args[0][1]).endswith("/moves/progress/host1")
+        assert result == expected_response
+
+    @patch("requests.Session.request")
+    def test_start_move_batch(self, mock_post):
+        hostnames = ["host1", "host2"]
+        expected_response = {"host1": 1, "host2": 2}
+        mock_response = Mock()
+        mock_response.json.return_value = expected_response
+        mock_post.return_value = mock_response
+
+        result = self.api.start_move_batch(hostnames)
+
+        mock_post.assert_called_once()
+        assert str(mock_post.call_args[0][1]).endswith("/moves/progress/batch")
+        assert mock_post.call_args[1]["json"] == {"hostnames": hostnames}
+        assert result == expected_response
+
+    @patch("requests.Session.request")
+    def test_update_move_status(self, mock_patch):
+        data = {"status": "ipmi_config", "message": "IPMI configured"}
+        expected_response = {"id": 1, "host": "host1", "status": "ipmi_config"}
+        mock_response = Mock()
+        mock_response.json.return_value = expected_response
+        mock_patch.return_value = mock_response
+
+        result = self.api.update_move_status(1, data)
+
+        mock_patch.assert_called_once()
+        assert str(mock_patch.call_args[0][1]).endswith("/moves/progress/1")
+        assert result == expected_response
+
+    @patch("requests.Session.request")
+    def test_get_move_status_error(self, mock_get):
+        mock_response = Mock()
+        mock_response.status_code = 500
+        mock_get.return_value = mock_response
+
+        with pytest.raises(APIServerException, match="Check the flask server logs"):
+            self.api.get_move_status("host1")
+
     @patch("requests.Session.request")
     def test_get_version(self, mock_get):
         expected_response = {"version": "1.0.0", "api_version": "2.0"}