Fetching Large Amounts of Data with Python Using the kintone Cursor API

Purpose

kintone supports REST APIs, making it easy to retrieve data and more.
However, the Get Records API is not suitable for retrieving more than 10,000 records.
Details regarding this can be found in the article below.
https://developer.cybozu.io/hc/ja/articles/360030757312

As stated in the above article, it is necessary to use the Cursor API when retrieving large amounts of data. This article provides sample code for retrieving large volumes of data using the Cursor API with Python.

Prerequisites

• Ability to perform basic Python operations
• Access to a kintone environment and an access token for using the REST API
  • "View" (Read) permission is sufficient
• Knowledge of basic kintone REST API operations
  • https://qiita.com/RyBB/items/daabb9b60d804ee2242f

About the Cursor API

Refer to the following:
https://developer.cybozu.io/hc/ja/articles/360029152012

The basic operations are as follows:

• POST to the Cursor API by specifying search criteria.
• Using the cursor ID from the POST result, all data does not exist [Incomplete in original]
• GET from the Cursor API using the cursor ID to retrieve record contents.
• Retrieve all records, or execute a DELETE on the Cursor API to terminate the cursor.

A created cursor can be explicitly deleted by either reading all the data or performing a DELETE.
Since there is a limit to the number of cursors that can be created, please be cautious when running processes in parallel.

Example of using the Cursor API with Python

By creating a class like the following, you can prevent forgetting to delete the cursor.

import dataclasses
import requests


@dataclasses.dataclass
class KintoneCursorParameter:
    app: int
    fields: list[str] = None
    query: str = None
    size: int = 100


class KintoneCursorError(Exception):
    pass


class KintoneCursor:
    url: str = ""
    headers: dict = {}
    finished: bool = False
    param: KintoneCursorParameter = None
    cursor_data: dict = None
    next: bool = False

    def __init__(self, base_url: str, api_token: str, param: KintoneCursorParameter):
        self.url = f"{base_url}/k/v1/records/cursor.json"
        self.headers = {"X-Cybozu-API-Token": api_token}
        self.param = param

    def __enter__(self):
        res = requests.post(
            self.url, headers=self.headers, json=dataclasses.asdict(self.param)
        )
        res_data = res.json()
        if res.status_code != 200:
            if "message" in res_data:
                raise KintoneCursorError(res_data["message"])
            raise KintoneCursorError(res.status_code)
        self.cursor_data = {"id": res_data["id"]}
        self.next = True

    def __exit__(self, exc_type, exc_value, traceback):
        if self.next:
            requests.delete(self.url, headers=self.headers, params=self.cursor_data)

    def get_record(self):
        while self.next:
            res = requests.get(self.url, headers=self.headers, params=self.cursor_data)
            res_data = res.json()
            if res.status_code != 200:
                if "message" in res_data:
                    raise KintoneCursorError(res_data["message"])
                raise KintoneCursorError(res.status_code)
            for record in res_data["records"]:
                yield record
            self.next = res_data["next"]

The usage of this class is as follows:

param = KintoneCursorParameter(
    app=2,
    size=2,
    fields=["レコード番号", "title", "count", "animal", "更新日時"],
    query='animal in ("犬")'
)
ctrl = KintoneCursor(
    "https://xxxxxxxx.cybozu.com", "APIトークン", param
)
with ctrl:
    for record in ctrl.get_record():
        print(record)

Remarks

It seems that other APIs can be operated from Python using pykintone.
https://github.com/icoxfog417/pykintone