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

qstash/asyncio/client.py

@@ -49,3 +49,27 @@ def __init__(
 
         self.dlq = AsyncDlqApi(self.http)
         """Dlq (Dead Letter Queue) api."""
+
+    async def liveness(self) -> str:
+        """
+        Check if the QStash API is alive.
+
+        :return: "OK" if the API is alive.
+        """
+        return await self.http.request(
+            path="/v2/liveness",
+            method="GET",
+            parse_response=False,
+        )
+
+    async def readiness(self) -> str:
+        """
+        Check if the QStash API is ready to accept requests.
+
+        :return: "OK" if the API is ready.
+        """
+        return await self.http.request(
+            path="/v2/readiness",
+            method="GET",
+            parse_response=False,
+        )

qstash/client.py

@@ -50,3 +50,27 @@ def __init__(
 
         self.dlq = DlqApi(self.http)
         """Dlq (Dead Letter Queue) api."""
+
+    def liveness(self) -> str:
+        """
+        Check if the QStash API is alive.
+
+        :return: "OK" if the API is alive.
+        """
+        return self.http.request(
+            path="/v2/liveness",
+            method="GET",
+            parse_response=False,
+        )
+
+    def readiness(self) -> str:
+        """
+        Check if the QStash API is ready to accept requests.
+
+        :return: "OK" if the API is ready.
+        """
+        return self.http.request(
+            path="/v2/readiness",
+            method="GET",
+            parse_response=False,
+        )

tests/asyncio/test_health.py

@@ -0,0 +1,15 @@
+import pytest
+
+from qstash import AsyncQStash
+
+
+@pytest.mark.asyncio
+async def test_liveness_async(async_client: AsyncQStash) -> None:
+    result = await async_client.liveness()
+    assert result == "OK"
+
+
+@pytest.mark.asyncio
+async def test_readiness_async(async_client: AsyncQStash) -> None:
+    result = await async_client.readiness()
+    assert result == "OK"

tests/test_health.py

@@ -0,0 +1,11 @@
+from qstash import QStash
+
+
+def test_liveness(client: QStash) -> None:
+    result = client.liveness()
+    assert result == "OK"
+
+
+def test_readiness(client: QStash) -> None:
+    result = client.readiness()
+    assert result == "OK"