📝

kintoneのカーソルAPIを使用してpythonで大量のデータを取得する

2022/01/29に公開約3,000字

目的

kintoneはREST APIをサポートしており、データの取得などが容易に行える。
しかし、レコードの取得APIは、1万件をこえるレコードを取得するには適していない。
この詳細については下記の記事で確認できる。

https://developer.cybozu.io/hc/ja/articles/360030757312

上記の記事にあるように大量のデータを取得する際には、カーソルAPIを用いる必要がある。
本記事ではPythonを使用してカーソルAPIで大量データを取得するためのサンプルコードを載せる。

前提

  • pythonの基本的な操作を行えること
  • kintoneの環境を有しており、RESTAPIを使用するためのアクセストークンを有していること
    • 権限は「読み取り」で良い
  • kintoneのRESTAPIの基本的な操作を知っていること

カーソルAPIについて

以下参照。

https://developer.cybozu.io/hc/ja/articles/360029152012

基本的な操作は以下の通りとなる。

  • 検索条件を指定してカーソルAPIをPOSTする。
  • POST結果のカーソルIDを使用して、全てのデータが存在しな
  • カーソルのIDを使用してカーソルAPIをGETし、レコードの内容を取得する
  • 全てのレコードを取得する、もしくは、カーソルAPIのDELETEを実行してカーソルを終了する。

作成したカーソルは全てのデータを読み取るか、DELETを行うことで明示的に削除することができる。
作成できるカーソルには上限があるので、並行実行の際には注意すること

pythonによるカーソルAPIの使用例

以下のようなクラスを作成することにより、カーソルの消し忘れを防止する。

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"]

このクラスの使用方法は以下の通りである

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)

備考

その他のAPIをpythonから操作するにはpykintoneで行える模様。

https://github.com/icoxfog417/pykintone

Discussion

ログインするとコメントできます