"""Lovelio API client for Python. Standard library only - no dependencies. Python 3.9+. Copy this file into your project or download it from https://lovelio.ai/sdk/lovelio.py Usage: from lovelio import Lovelio client = Lovelio(api_key=os.environ["LOVELIO_API_KEY"]) jobs = client.jobs.list(status="active") job = client.jobs.get(jobs["data"][0]["id"]) # Async creates return a task - wait for it: created = client.jobs.create(title="Property Manager", client_id="cli_...") client.wait_for_task(created["task_id"]) # Page through everything: for candidate in client.candidates.iterate(): ... # Verify a webhook delivery: ok = verify_webhook_signature(payload, signature, timestamp, secret) """ from __future__ import annotations import hashlib import hmac import json import time import urllib.error import urllib.parse import urllib.request import uuid from typing import Any, Dict, Iterator, List, Optional DEFAULT_BASE_URL = "https://lovelio.ai/api/v1" class LovelioError(Exception): """A non-2xx API response. Carries code, status and request_id.""" def __init__(self, code: str, message: str, status: int, request_id: Optional[str] = None): super().__init__(f"{code}: {message}") self.code = code self.message = message self.status = status self.request_id = request_id class _Resource: """A standard resource: list / iterate / get.""" def __init__(self, client: "Lovelio", path: str): self._client = client self._path = path def list(self, **params: Any) -> Dict[str, Any]: """Returns {"data": [...], "meta": {...}} with cursor pagination meta.""" return self._client.request("GET", self._path, params=params) def iterate(self, **params: Any) -> Iterator[Dict[str, Any]]: """Follow meta.next_cursor until the list is exhausted.""" cursor: Optional[str] = None while True: page = self.list(**{**params, **({"after": cursor} if cursor else {})}) for row in page.get("data", []): yield row meta = page.get("meta") or {} if not meta.get("has_more") or not meta.get("next_cursor"): return cursor = meta["next_cursor"] def get(self, resource_id: str) -> Dict[str, Any]: return self._client.request_data("GET", f"{self._path}/{resource_id}") class _WritableResource(_Resource): def create(self, **body: Any) -> Dict[str, Any]: return self._client.request_data("POST", self._path, body=body) def update(self, resource_id: str, **body: Any) -> Dict[str, Any]: return self._client.request_data("PATCH", f"{self._path}/{resource_id}", body=body) def delete(self, resource_id: str) -> None: self._client.request("DELETE", f"{self._path}/{resource_id}") class Lovelio: def __init__(self, api_key: str, base_url: str = DEFAULT_BASE_URL, max_retries: int = 2, timeout: float = 30.0): if not api_key: raise ValueError("api_key is required") self._api_key = api_key self._base_url = base_url.rstrip("/") self._max_retries = max_retries self._timeout = timeout self.jobs = _WritableResource(self, "/jobs") self.candidates = _WritableResource(self, "/candidates") self.applications = _WritableResource(self, "/applications") self.clients = _WritableResource(self, "/clients") self.interviews = _WritableResource(self, "/interviews") self.placements = _WritableResource(self, "/placements") self.submissions = _WritableResource(self, "/submissions") self.talent_pools = _WritableResource(self, "/talent-pools") self.webhooks = _WritableResource(self, "/webhooks") self.activities = _Resource(self, "/activities") self.bd_targets = _Resource(self, "/bd/targets") self.bd_leads = _Resource(self, "/bd/leads") # -- transport ---------------------------------------------------------- def request(self, method: str, path: str, params: Optional[Dict[str, Any]] = None, body: Optional[Dict[str, Any]] = None) -> Any: """Raw request. Raises LovelioError on any non-2xx. List responses come back as {"data": [...], "meta": {...}}; use request_data() to unwrap single-resource responses.""" url = self._base_url + path if params: clean = {k: str(v) for k, v in params.items() if v is not None} if clean: url += "?" + urllib.parse.urlencode(clean) headers = { "Authorization": f"Bearer {self._api_key}", "User-Agent": "lovelio-sdk-py/1.0", } data = None if body is not None: headers["Content-Type"] = "application/json" data = json.dumps(body).encode() # Every POST carries an idempotency key - the API requires it on # mutating POSTs, and a random key per call is always safe. if method == "POST": headers["Idempotency-Key"] = str(uuid.uuid4()) last_error: Optional[LovelioError] = None for attempt in range(self._max_retries + 1): req = urllib.request.Request(url, method=method, headers=headers, data=data) try: with urllib.request.urlopen(req, timeout=self._timeout) as res: if res.status == 204: return None payload = json.loads(res.read().decode() or "null") if payload and isinstance(payload.get("data"), list): return {"data": payload["data"], "meta": payload.get("meta")} return payload except urllib.error.HTTPError as e: try: err = json.loads(e.read().decode() or "{}") except (ValueError, AttributeError): err = {} error = (err.get("error") or {}) meta = (err.get("meta") or {}) last_error = LovelioError( error.get("code", "UNKNOWN"), error.get("message", f"HTTP {e.code}"), e.code, meta.get("request_id"), ) # Retry 429 and 5xx with backoff; everything else is the caller's problem. if e.code != 429 and e.code < 500: raise last_error from None if attempt < self._max_retries: retry_after = e.headers.get("Retry-After") if e.headers else None delay = float(retry_after) if retry_after and retry_after.isdigit() else 0.5 * (2 ** attempt) time.sleep(delay) raise last_error or LovelioError("UNKNOWN", "Request failed", 0) def request_data(self, method: str, path: str, params: Optional[Dict[str, Any]] = None, body: Optional[Dict[str, Any]] = None) -> Any: """Like request(), but unwraps the envelope to data.""" res = self.request(method, path, params=params, body=body) if isinstance(res, dict) and "data" in res and not isinstance(res["data"], list): return res["data"] return res.get("data", res) if isinstance(res, dict) else res # -- conveniences -------------------------------------------------------- def batch(self, operations: List[Dict[str, Any]]) -> Dict[str, Any]: """Run batch operations: [{"op": "add_note", "payload": {...}}, ...]""" return self.request_data("POST", "/batch", body={"operations": operations}) def search_candidates(self, query: str, **extra: Any) -> Dict[str, Any]: return self.request_data("POST", "/candidates/search", body={"query": query, **extra}) def bd_patch(self) -> Dict[str, Any]: """The BD patch: the agency's declared territory in its own words.""" return self.request_data("GET", "/bd/patch") def webhook_events(self) -> Dict[str, Any]: """The machine-readable webhook event catalogue.""" return self.request_data("GET", "/webhooks/events") def wait_for_task(self, task_id: str, timeout: float = 120.0, interval: float = 2.0) -> Dict[str, Any]: """Poll an async task (202 responses return task_id) until it completes or fails. Returns the task's final state.""" deadline = time.monotonic() + timeout while True: task = self.request_data("GET", f"/tasks/{task_id}") if task["status"] == "completed": return task if task["status"] == "failed": raise LovelioError("TASK_FAILED", f"Task {task_id} failed", 200) if time.monotonic() >= deadline: raise LovelioError("TASK_TIMEOUT", f"Task {task_id} still {task['status']} after {timeout}s", 200) time.sleep(interval) def verify_webhook_signature(payload: str, signature: str, timestamp: str, secret: str, tolerance_seconds: int = 300) -> bool: """Verify a webhook delivery. Lovelio signs every delivery with HMAC-SHA256 over f"{timestamp}.{raw_body}" and sends it as X-Lovelio-Signature: sha256= plus X-Lovelio-Timestamp. Always verify against the RAW request body, before parsing. """ try: ts = float(timestamp) except (TypeError, ValueError): return False if abs(time.time() - ts) > tolerance_seconds: return False expected = "sha256=" + hmac.new(secret.encode(), f"{timestamp}.{payload}".encode(), hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature)