SDK для Python

import os

from soroka.sdk import Soroka, SorokaError

async with Soroka("https://soroka.tools", key=os.environ["API_KEY"]) as s:
    me = await s.me()
    posts = await s.tg.posts("@telegram", limit=20)

    job = await s.jobs.create("parser.preset", {"preset": "tg.channel_posts",
                                                "params": {"ref": "@telegram", "limit": 300}})
    card = await s.jobs.wait(job["job_id"])          # опрашивает карточку до done/failed/cancelled
    csv = await s.jobs.result(job["job_id"], "csv")  # bytes

    sub = await s.listen.subscribe(["@telegram"], events=["post"])
    async for ev in sub.stream():                    # сам переподключается с Last-Event-ID
        print(ev["id"], ev["kind"])

import os
import time
import uuid

import httpx


class ApiError(Exception):
    def __init__(self, status: int, body: dict):
        err = body.get("error") or {}
        super().__init__(f"{status} {err.get('kind')}/{err.get('code')}: {err.get('message')}")
        self.status, self.kind, self.code = status, err.get("kind"), err.get("code")
        self.retry_after, self.request_id = err.get("retry_after"), body.get("request_id")


class Client:
    def __init__(self, base: str = "https://soroka.tools", key: str | None = None, **http_kw):
        self.http = httpx.Client(base_url=base.rstrip("/") + "/v1", timeout=60,
                                 headers={"Authorization": f"Bearer {key or os.environ['API_KEY']}"},
                                 **http_kw)

    def request(self, method: str, path: str, *, idempotent: bool = False, retries: int = 3, **kw):
        headers = dict(kw.pop("headers", {}))
        if idempotent:                       # один ключ на все повторы: действие не выполнится дважды
            headers.setdefault("Idempotency-Key", str(uuid.uuid4()))
        for attempt in range(retries + 1):
            r = self.http.request(method, path, headers=headers, **kw)
            if r.status_code < 400:
                return r
            body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
            err = ApiError(r.status_code, body)
            # повторяем только то, что лечится ожиданием, и только короткое ожидание
            if attempt < retries and err.retry_after is not None and err.retry_after <= 60:
                time.sleep(max(1, err.retry_after))
                continue
            raise err

    def get(self, path: str, **params) -> dict:
        return self.request("GET", path, params=params).json()

    def post(self, path: str, body: dict | None = None, *, idempotent: bool = False) -> dict:
        return self.request("POST", path, json=body or {}, idempotent=idempotent).json()

    def pages(self, path: str, **params):
        """Все элементы списка по курсору."""
        while True:
            page = self.get(path, **params)
            yield from page["items"]
            if not page.get("next_cursor"):
                return
            params["cursor"] = page["next_cursor"]

    def run_job(self, body: dict, *, fmt: str = "json", every: float = 5.0) -> bytes:
        """Поставить задачу, дождаться и скачать результат."""
        job_id = self.post("/jobs", body, idempotent=True)["job_id"]
        while (card := self.get(f"/jobs/{job_id}"))["status"] in ("queued", "running"):
            time.sleep(every)
        if card["status"] != "done":
            raise RuntimeError(f"задача {job_id}: {card['status']} {card.get('error') or ''}")
        return self.request("GET", f"/jobs/{job_id}/result", params={"format": fmt}).content

from client import ApiError, Client

api = Client()
print(api.get("/me")["scopes"])

posts = api.post("/tg/posts", {"chat": "@telegram", "limit": 20})["items"]

for job in api.pages("/jobs", status="done"):
    print(job["job_id"], job["result_summary"])

data = api.run_job({"kind": "parser.preset",
                    "params": {"preset": "tg.channel_posts",
                               "params": {"ref": "@telegram", "limit": 300}}}, fmt="csv")
open("posts.csv", "wb").write(data)

try:
    api.post("/tg/posts", {"chat": "@no_such_channel_xyz"})
except ApiError as e:
    print(e.kind, e.code, e.request_id)       # no_data UsernameNotOccupied …