📝
kintoneのカーソルAPIを使用してpythonで大量のデータを取得する
目的
kintoneはREST APIをサポートしており、データの取得などが容易に行える。
しかし、レコードの取得APIは、1万件をこえるレコードを取得するには適していない。
この詳細については下記の記事で確認できる。
上記の記事にあるように大量のデータを取得する際には、カーソルAPIを用いる必要がある。
本記事ではPythonを使用してカーソルAPIで大量データを取得するためのサンプルコードを載せる。
前提
- pythonの基本的な操作を行えること
- kintoneの環境を有しており、RESTAPIを使用するためのアクセストークンを有していること
- 権限は「読み取り」で良い
- kintoneのRESTAPIの基本的な操作を知っていること
カーソルAPIについて
以下参照。
基本的な操作は以下の通りとなる。
- 検索条件を指定してカーソル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で行える模様。
Discussion