INNER CODE UNIT · Python

get

defiprime/defiprime1 · scripts/algolia-index.py:148

    def get(self, path):
        return self.http("GET", self.url(path), self.headers, None)

    def wait(self, index, task_id, sleep=time.sleep):
        for _ in range(TASK_POLL_LIMIT):
            status = self.get(f"/1/indexes/{index}/task/{task_id}").get("status")
            if status == "published":
                return
            sleep(TASK_POLL_SECONDS)
        raise AlgoliaError(f"task {task_id} on {index} never published")

    def operation(self, source, operation, destination, scope=None):
        payload = {"operation": operation, "destination": destination}
        if scope is not None:
            payload["scope"] = scope
        response = self.post(
            f"/1/indexes/{source}/operation",
            json.dumps(payload, separators=(",", ":")).encode("utf-8"),

View source record →