Task results

import { setTimeout as delay } from "node:timers/promises";

const baseUrl = "https://app.lev8.com";
const apiKey = process.env.LEV8_API_KEY;

async function lev8Json(path, options = {}) {
  const response = await fetch(`${baseUrl}${path}`, {
    ...options,
    headers: {
      Accept: "application/json",
      Authorization: `Bearer ${apiKey}`,
      ...options.headers,
    },
  });
  const requestId = response.headers.get("x-request-id");
  const body = await response.json();
  if (!response.ok) {
    throw new Error(
      body.error?.message ??
        `lev8 request failed with HTTP ${response.status} (request_id=${requestId ?? "unknown"})`,
    );
  }
  return { body, requestId };
}

const created = await lev8Json("/v1/entity-search/create_task", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    "Idempotency-Key": crypto.randomUUID(),
  },
  body: JSON.stringify({
    entity_type: "company",
    objective: "Developer tooling companies in Europe with recent seed funding",
    limit: 20,
    enrich_fields: ["website", "funding stage"],
  }),
});

const leadsSearchId = created.body.leads_search_id;
const entities = [];
let pts = 0;
let terminalStatus = null;

while (true) {
  const statusUrl = new URL("/v1/entity-search/status", baseUrl);
  statusUrl.searchParams.set("leads_search_id", leadsSearchId);
  const { body: statusBody } = await lev8Json(`${statusUrl.pathname}${statusUrl.search}`);
  const task = statusBody.status;

  while (pts < task.ready_count) {
    const fetchUrl = new URL("/v1/entity-search/fetch", baseUrl);
    fetchUrl.search = new URLSearchParams({
      leads_search_id: leadsSearchId,
      pts: String(pts),
      num: "100",
    });
    const { body: page } = await lev8Json(`${fetchUrl.pathname}${fetchUrl.search}`);
    if (page.count === 0) throw new Error("lev8 returned an empty page before ready_count");
    entities.push(...page.entities);
    pts += page.count;
  }

  if (task.status === "done" || task.status === "error") {
    terminalStatus = task;
    break;
  }
  await delay(5000);
}

console.log({ leadsSearchId, terminalStatus, entities });

import asyncio
import os
import uuid
import httpx


BASE_URL = "https://app.lev8.com"
API_KEY = os.environ["LEV8_API_KEY"]


async def lev8_json(client, method, path, **kwargs):
    headers = {
        "Accept": "application/json",
        "Authorization": f"Bearer {API_KEY}",
        **kwargs.pop("headers", {}),
    }
    response = await client.request(method, path, headers=headers, **kwargs)
    request_id = response.headers.get("x-request-id")
    body = response.json()
    if response.is_error:
        message = body.get("error", {}).get("message")
        raise RuntimeError(
            message
            or f"lev8 request failed with HTTP {response.status_code} "
            f"(request_id={request_id or 'unknown'})"
        )
    return body, request_id


async def entity_search():
    timeout = httpx.Timeout(connect=10, read=60, write=30, pool=10)
    async with httpx.AsyncClient(base_url=BASE_URL, timeout=timeout) as client:
        created, request_id = await lev8_json(
            client,
            "POST",
            "/v1/entity-search/create_task",
            headers={"Idempotency-Key": str(uuid.uuid4())},
            json={
                "entity_type": "company",
                "objective": "Developer tooling companies in Europe with recent seed funding",
                "limit": 20,
                "enrich_fields": ["website", "funding stage"],
            },
        )
        leads_search_id = created["leads_search_id"]
        entities = []
        pts = 0

        while True:
            status_body, _ = await lev8_json(
                client,
                "GET",
                "/v1/entity-search/status",
                params={"leads_search_id": leads_search_id},
            )
            task = status_body["status"]

            while pts < task["ready_count"]:
                page, _ = await lev8_json(
                    client,
                    "GET",
                    "/v1/entity-search/fetch",
                    params={
                        "leads_search_id": leads_search_id,
                        "pts": pts,
                        "num": 100,
                    },
                )
                if page["count"] == 0:
                    raise RuntimeError("lev8 returned an empty page before ready_count")
                entities.extend(page["entities"])
                pts += page["count"]

            if task["status"] in {"done", "error"}:
                return {
                    "request_id": request_id,
                    "leads_search_id": leads_search_id,
                    "terminal_status": task,
                    "entities": entities,
                }
            await asyncio.sleep(5)


if __name__ == "__main__":
    print(asyncio.run(entity_search()))