42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
|
"""JobTread Pave API helper. Reads creds from .jobtread-api.json."""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import urllib.request
|
||
|
|
|
||
|
|
PAVE_URL = "https://api.jobtread.com/pave"
|
||
|
|
_creds = None
|
||
|
|
|
||
|
|
def _get_creds():
|
||
|
|
global _creds
|
||
|
|
if _creds:
|
||
|
|
return _creds
|
||
|
|
paths = [
|
||
|
|
os.environ.get("JOBTREAD_CREDS"),
|
||
|
|
os.path.join(os.path.dirname(__file__), "..", "data", ".jobtread-api.json"),
|
||
|
|
"/home/node/.openclaw/workspace/agents/tom/.jobtread-api.json",
|
||
|
|
]
|
||
|
|
for p in paths:
|
||
|
|
if p and os.path.exists(p):
|
||
|
|
with open(p) as f:
|
||
|
|
_creds = json.load(f)
|
||
|
|
return _creds
|
||
|
|
raise RuntimeError("No .jobtread-api.json found")
|
||
|
|
|
||
|
|
def get_grant_key():
|
||
|
|
return _get_creds()["grantKey"]
|
||
|
|
|
||
|
|
def get_org_id():
|
||
|
|
return _get_creds()["organizationId"]
|
||
|
|
|
||
|
|
def pave_query(query: dict) -> dict:
|
||
|
|
"""Execute a Pave API query. Injects grantKey automatically."""
|
||
|
|
payload = {"query": {"$": {"grantKey": get_grant_key()}, **query}}
|
||
|
|
data = json.dumps(payload).encode()
|
||
|
|
req = urllib.request.Request(PAVE_URL, data=data,
|
||
|
|
headers={"Content-Type": "application/json"}, method="POST")
|
||
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
|
|
result = json.loads(resp.read())
|
||
|
|
if "errors" in result:
|
||
|
|
raise RuntimeError(f"Pave API error: {result['errors']}")
|
||
|
|
return result
|