v0.0.4b for exospherehost pypi sdk

python-sdk/README.md

@@ -1,2 +1,24 @@
 # ExosphereHost Python SDK
-This SDK is official python SDK for ExosphereHost and interacting with exospherehost.
+This is the official Python SDK for ExosphereHost and for interacting with ExosphereHost.
+
+## Node Creation
+You can simply connect to exosphere state manager and start creating your nodes, as shown in sample below: 
+
+```python
+from exospherehost import Runtime, BaseNode
+from typing import Any
+import os
+
+class SampleNode(BaseNode):
+    async def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
+        print(inputs)
+        return {"message": "success"}
+
+runtime = Runtime("SampleNamespace", os.getenv("EXOSPHERE_STATE_MANAGER_URI", "http://localhost:8000"), os.getenv("EXOSPHERE_API_KEY", ""))
+
+runtime.connect([SampleNode()])
+runtime.start()
+```
+
+## Support
+For first-party support and questions, do not hesitate to reach out to us at <nivedit@exosphere.host>.

python-sdk/exospherehost/__init__.py

@@ -1,6 +1,7 @@
-from ._version import VERSION
+from ._version import version as __version__
+from .runtime import Runtime
+from .node.BaseNode import BaseNode
 
-__version__ = VERSION
+VERSION = __version__
 
-def test():
-    print(f"ExosphereHost PySDK v{VERSION}")
+__all__ = ["Runtime", "BaseNode", "VERSION"]

python-sdk/exospherehost/_version.py

@@ -1 +1 @@
-VERSION = "0.0.3b"
+version = "0.0.4b"

python-sdk/exospherehost/node/BaseNode.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import Optional
+from typing import Optional, Any
 
 
 class BaseNode(ABC):
@@ -8,7 +8,7 @@ def __init__(self, unique_name: Optional[str] = None):
         self.unique_name: Optional[str] = unique_name
 
     @abstractmethod
-    async def execute(self):
+    async def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
         pass
 
     def get_unique_name(self) -> str:

python-sdk/exospherehost/node/__init__.py

@@ -1 +0,0 @@
-from .BaseNode import BaseNode

python-sdk/exospherehost/runtime.py

@@ -1,14 +1,17 @@
 import asyncio
 from asyncio import Queue, sleep
-import logging
 from typing import Any, List
-from .node import BaseNode
+from .node.BaseNode import BaseNode
 from aiohttp import ClientSession
+from logging import getLogger
+
+logger = getLogger(__name__)
 
 class Runtime:
 
-    def __init__(self, namespace: str, state_manager_uri: str, batch_size: int = 16, workers=4, state_manage_version: int = 0, poll_interval: int = 10):
+    def __init__(self, namespace: str, state_manager_uri: str, key: str, batch_size: int = 16, workers=4, state_manage_version: str = "v0", poll_interval: int = 1):
         self._namespace = namespace
+        self._key = key
         self._batch_size = batch_size
         self._connected = False
         self._state_queue = Queue(maxsize=2*batch_size)
@@ -34,33 +37,61 @@ def _get_executed_endpoint(self, state_id: str):
     def _get_errored_endpoint(self, state_id: str):
         return f"{self._state_manager_uri}/{str(self._state_manager_version)}/namespace/{self._namespace}/states/{state_id}/errored"
 
-    async def connect(self, nodes: List[BaseNode]):
+    def connect(self, nodes: List[BaseNode]):
         self._nodes = self._validate_nodes(nodes)
         self._node_names = [node.get_unique_name() for node in nodes]
         self._node_mapping = {node.get_unique_name(): node for node in self._nodes}
         self._connected = True
 
     async def _enqueue_call(self):
         async with ClientSession() as session:
-            async with session.post(self._get_enque_endpoint(), json={"nodes": self._node_names, "batch_size": self._batch_size}) as response:
-                return await response.json()
+            endpoint = self._get_enque_endpoint()
+            body = {"nodes": self._node_names, "batch_size": self._batch_size}
+            headers = {"x-api-key": self._key}
+
+            async with session.post(endpoint, json=body, headers=headers) as response:
+                res = await response.json()
+
+                if response.status != 200:
+                    logger.error(f"Failed to enqueue states: {res}")
+
+                return res
 
     async def _enqueue(self):
-        if self._state_queue.qsize() < self._batch_size: 
-            data = await self._enqueue_call()
-            for state in data["states"]:
-                await self._state_queue.put(state)
-        await sleep(self._poll_interval)
+        while True:
+            try:
+                if self._state_queue.qsize() < self._batch_size: 
+                    data = await self._enqueue_call()
+                    for state in data["states"]:
+                        await self._state_queue.put(state)
+            except Exception as e:
+                logger.error(f"Error enqueuing states: {e}")
+
+            await sleep(self._poll_interval)
 
     async def _notify_executed(self, state_id: str, outputs: dict[str, Any]):
         async with ClientSession() as session:
-            async with session.post(self._get_executed_endpoint(state_id), json={"outputs": outputs}) as response:
-                return await response.json()
+            endpoint = self._get_executed_endpoint(state_id)
+            body = {"outputs": outputs}
+            headers = {"x-api-key": self._key}
+
+            async with session.post(endpoint, json=body, headers=headers) as response:
+                res = await response.json()
+
+                if response.status != 200:
+                    logger.error(f"Failed to notify executed state {state_id}: {res}")
 
     async def _notify_errored(self, state_id: str, error: str):
         async with ClientSession() as session:
-            async with session.post(self._get_errored_endpoint(state_id), json={"error": error}) as response:
-                return await response.json()
+            endpoint = self._get_errored_endpoint(state_id)
+            body = {"error": error}
+            headers = {"x-api-key": self._key}
+
+            async with session.post(endpoint, json=body, headers=headers) as response:
+                res =  await response.json()
+
+                if response.status != 200:
+                    logger.error(f"Failed to notify errored state {state_id}: {res}")
 
     def _validate_nodes(self, nodes: List[BaseNode]):
         invalid_nodes = []
@@ -81,15 +112,24 @@ async def _worker(self):
             try:
                 node = self._node_mapping[state["node_name"]]
                 outputs = await node.execute(state["inputs"]) # type: ignore
-                await self._notify_executed(state["id"], outputs)
+                await self._notify_executed(state["state_id"], outputs)
             except Exception as e:
-                await self._notify_errored(state["id"], str(e))
+                await self._notify_errored(state["state_id"], str(e))
+
+            self._state_queue.task_done() # type: ignore
 
-    async def start(self):
+    async def _start(self):
         if not self._connected:
             raise RuntimeError("Runtime not connected, you need to call Runtime.connect() before calling Runtime.start()")
 
         poller = asyncio.create_task(self._enqueue())
         worker_tasks = [asyncio.create_task(self._worker()) for _ in range(self._workers)]
 
-        await asyncio.gather(poller, *worker_tasks)
+        await asyncio.gather(poller, *worker_tasks)
+
+    def start(self):
+        try:
+            loop = asyncio.get_running_loop()
+            return loop.create_task(self._start())
+        except RuntimeError:
+            asyncio.run(self._start())

python-sdk/pyproject.toml

@@ -25,7 +25,7 @@ Documentation = "https://docs.exosphere.host"
 Repository = "https://github.com/exospherehost/exospherehost/tree/main/python-sdk"
 
 [tool.setuptools.dynamic]
-version = {attr = "exospherehost.__version__"}
+version = {attr = "exospherehost._version.version"}
 
 [dependency-groups]
 dev = [

python-sdk/sample.py

@@ -0,0 +1,13 @@
+from exospherehost import Runtime, BaseNode
+from typing import Any
+import os
+
+class SampleNode(BaseNode):
+    async def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
+        print(inputs)
+        return {"message": "success"}
+
+runtime = Runtime("SampleNamespace", os.getenv("EXOSPHERE_STATE_MANAGER_URI", "http://localhost:8000"), os.getenv("EXOSPHERE_API_KEY", ""))
+
+runtime.connect([SampleNode()])
+runtime.start()