INNER CODE UNIT · Python

post

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

    def post(self, path, payload):
        return self.http("POST", self.url(path), self.headers, payload)

    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

View source record →