[ardabot] SDK changes for upstash/qstash-server#884

qstash/asyncio/client.py

@@ -6,6 +6,7 @@
 from qstash.asyncio.http import AsyncHttpClient
 from qstash.asyncio.message import AsyncMessageApi
 from qstash.asyncio.queue import AsyncQueueApi
+from qstash.asyncio.readiness import AsyncReadinessApi
 from qstash.asyncio.schedule import AsyncScheduleApi
 from qstash.asyncio.signing_key import AsyncSigningKeyApi
 from qstash.asyncio.url_group import AsyncUrlGroupApi
@@ -49,3 +50,6 @@ def __init__(
 
         self.dlq = AsyncDlqApi(self.http)
         """Dlq (Dead Letter Queue) api."""
+
+        self.readiness = AsyncReadinessApi(self.http)
+        """Readiness api."""

qstash/asyncio/readiness.py

@@ -0,0 +1,18 @@
+from qstash.asyncio.http import AsyncHttpClient
+from qstash.readiness import ReadinessResponse, parse_readiness_response
+
+
+class AsyncReadinessApi:
+    def __init__(self, http: AsyncHttpClient) -> None:
+        self._http = http
+
+    async def check(self) -> ReadinessResponse:
+        """
+        Checks the readiness of the QStash service.
+        """
+        response = await self._http.request(
+            path="/v2/readiness",
+            method="GET",
+        )
+
+        return parse_readiness_response(response)

qstash/client.py

@@ -6,6 +6,7 @@
 from qstash.http import RetryConfig, HttpClient
 from qstash.message import MessageApi
 from qstash.queue import QueueApi
+from qstash.readiness import ReadinessApi
 from qstash.schedule import ScheduleApi
 from qstash.signing_key import SigningKeyApi
 from qstash.url_group import UrlGroupApi
@@ -50,3 +51,6 @@ def __init__(
 
         self.dlq = DlqApi(self.http)
         """Dlq (Dead Letter Queue) api."""
+
+        self.readiness = ReadinessApi(self.http)
+        """Readiness api."""

qstash/readiness.py

@@ -0,0 +1,32 @@
+import dataclasses
+from typing import Any, Dict
+
+from qstash.http import HttpClient
+
+
+@dataclasses.dataclass
+class ReadinessResponse:
+    ready: bool
+    """Whether the QStash service is ready to accept requests."""
+
+
+def parse_readiness_response(response: Dict[str, Any]) -> ReadinessResponse:
+    return ReadinessResponse(
+        ready=response["ready"],
+    )
+
+
+class ReadinessApi:
+    def __init__(self, http: HttpClient) -> None:
+        self._http = http
+
+    def check(self) -> ReadinessResponse:
+        """
+        Checks the readiness of the QStash service.
+        """
+        response = self._http.request(
+            path="/v2/readiness",
+            method="GET",
+        )
+
+        return parse_readiness_response(response)

tests/asyncio/test_readiness.py

@@ -0,0 +1,9 @@
+import pytest
+
+from qstash import AsyncQStash
+
+
+@pytest.mark.asyncio
+async def test_check_async(async_client: AsyncQStash) -> None:
+    response = await async_client.readiness.check()
+    assert response.ready is True

tests/test_readiness.py

@@ -0,0 +1,6 @@
+from qstash import QStash
+
+
+def test_check(client: QStash) -> None:
+    response = client.readiness.check()
+    assert response.ready is True