feat, fix: communicating to host + timeout handle

Author: shreykx

genstack/__init__.py

@@ -0,0 +1,3 @@
+from .genstack import Genstack
+
+__all__ = ["Genstack"]

genstack/genstack.py

@@ -0,0 +1,44 @@
+import httpx
+from typing import Dict, Any, Optional, Union
+import asyncio
+
+class Genstack:
+    def __init__(self, api_key: str, base_url : Optional[str] = None):
+        if not api_key.startswith("gen-") or any(c.isspace() for c in api_key):
+            raise ValueError("API key must start with 'gen-' and contain no spaces or line breaks.")
+        self.api_key : str = api_key
+        self.base_url : str = base_url or "http://localhost:8000"
+    async def __call(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
+            try:
+                response = await client.post(
+                    url=f"{self.base_url}/api/v1/sdk/generate",
+                    headers={
+                        "x-api-key" : f"{self.api_key}"
+                    },
+                    json=payload
+                )
+                response.raise_for_status()
+                return response.json()
+            except httpx.HTTPStatusError as e:
+                return e.response.json()
+            except Exception as e:
+                return {"error" : str(e)}
+    async def __generate_async(self, payload : Dict[str, Any]) -> Dict[str, Any]:
+        return await self.__call(payload=payload)
+    def generate(self, input : Union[str, Dict[str, Any]], model : Optional[str] = "auto", track : Optional[str] = None) -> Dict[str, Any]:
+        if not track :
+            raise ValueError("Track is required.")
+        if not isinstance(input, (str, dict)):
+            raise TypeError("Payload must be a string or a dictionary.")
+        if isinstance(input, str):
+            inner_payload = {"input": input}
+        else:
+            inner_payload = input
+
+        payload = {
+            "payload": inner_payload,
+            "model": model,
+            "track": track
+        }
+        return asyncio.run(self.__generate_async(payload=payload))

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "genstack"
-version = "0.1.5"
+version = "0.1.6"
 description = "Universal AI SDK from Genstack"
 authors = [{name="Shrey Kumar", email="shreyk.dev@gmail.com"}]
 readme = "README.md"
@@ -12,3 +12,5 @@ dependencies = ["httpx"]
 requires = ["hatchling"]
 build-backend = "hatchling.build"
 
+[tool.pytest.ini_options]
+pythonpath = ["."]

requirements.txt

@@ -0,0 +1,17 @@
+from genstack import Genstack
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+
+
+client = Genstack(api_key=os.getenv("GENSTACK_API_KEY"))
+
+
+res = client.generate(input="3 fun facts about Ferrari", track="dragon", model="gpt-4-1-nano-oai")
+
+if "output" in res:
+    print(res["output"][0]["output"]["text"])
+else:
+    print("Error:", res.get("message", "Unknown error"))

tests/call_test.py

@@ -0,0 +1,7 @@
+from genstack import Genstack
+import pytest
+
+def test_generate_with_invalid_type():
+    client = Genstack(api_key="gen-")
+    with pytest.raises(TypeError):
+        client.generate(input=123, track="test-track")

tests/test_generate.py

@@ -0,0 +1,10 @@
+class Sample:
+    def __init__(self) -> None:
+        timeout = 30
+    @staticmethod
+    def say(self) -> None:
+        print(self.timeout)
+        
+sam = Sample()
+
+sam.say()