jobyssey/tools/twenty_client.py

"""Minimal Twenty CRM REST client (stdlib only).

Shared by tools/twenty_enrich.py and tools/gather_recent_context.py.

Auth: set TWENTY_API_KEY (workspace API key from Settings -> API & Webhooks)
and optionally TWENTY_BASE_URL (default: https://crm.prad.nu).

Both consumer scripts also run keyless: they compute a *plan* JSON from local
data plus an optional CRM snapshot, and the /sync-twenty or
/gather-recent-context command applies the plan through the Twenty MCP
server, which is session-authenticated via OAuth. The REST path exists so the
whole loop can also run headless (cron, CI) with an API key.
"""
from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request

DEFAULT_BASE_URL = "https://crm.prad.nu"
PAGE_LIMIT = 60  # Twenty REST caps page size at 60


class TwentyError(RuntimeError):
    pass


class TwentyClient:
    def __init__(self, base_url: str | None = None, api_key: str | None = None):
        self.base_url = (base_url or os.environ.get("TWENTY_BASE_URL") or DEFAULT_BASE_URL).rstrip("/")
        self.api_key = api_key or os.environ.get("TWENTY_API_KEY")
        if not self.api_key:
            raise TwentyError(
                "TWENTY_API_KEY is not set. Either export it, or run without --apply "
                "to emit a plan JSON and apply it through the Twenty MCP session."
            )

    def _request(self, method: str, path: str, body: dict | None = None,
                 params: dict | None = None, retries: int = 3):
        url = f"{self.base_url}{path}"
        if params:
            url += "?" + urllib.parse.urlencode(params)
        data = json.dumps(body).encode() if body is not None else None
        last_err = None
        for attempt in range(retries):
            req = urllib.request.Request(url, data=data, method=method, headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
                "Accept": "application/json",
            })
            try:
                with urllib.request.urlopen(req, timeout=60) as resp:
                    return json.loads(resp.read().decode())
            except urllib.error.HTTPError as e:  # 4xx never retried: same input, same answer
                detail = e.read().decode(errors="replace")[:500]
                if e.code == 429 or e.code >= 500:
                    last_err = TwentyError(f"{method} {path} -> {e.code}: {detail}")
                    time.sleep(1.5 * (attempt + 1))
                    continue
                raise TwentyError(f"{method} {path} -> {e.code}: {detail}") from e
            except urllib.error.URLError as e:
                last_err = TwentyError(f"{method} {path} -> {e.reason}")
                time.sleep(1.5 * (attempt + 1))
        raise last_err

    # -- records ------------------------------------------------------------

    def list_all(self, plural: str, order_by: str = "id") -> list[dict]:
        """Every record of an object, cursor-paged."""
        out: list[dict] = []
        cursor = None
        while True:
            params = {"limit": PAGE_LIMIT, "order_by": order_by}
            if cursor:
                params["starting_after"] = cursor
            payload = self._request("GET", f"/rest/{plural}", params=params)
            page = payload.get("data", {}).get(plural, [])
            out.extend(page)
            page_info = payload.get("pageInfo", {})
            if not page_info.get("hasNextPage") or not page:
                break
            cursor = page_info.get("endCursor") or page[-1]["id"]
        return out

    def create_batch(self, plural: str, records: list[dict]) -> list[dict]:
        """Create records in chunks of PAGE_LIMIT via the batch endpoint."""
        created: list[dict] = []
        for i in range(0, len(records), PAGE_LIMIT):
            payload = self._request("POST", f"/rest/batch/{plural}",
                                    body=records[i:i + PAGE_LIMIT])
            created.extend(payload.get("data", {}).get(plural, []))
        return created

    def update_one(self, plural: str, record_id: str, patch: dict) -> dict:
        payload = self._request("PATCH", f"/rest/{plural}/{record_id}", body=patch)
        data = payload.get("data", {})
        return next(iter(data.values())) if data else payload