LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@sato_7411(Kazunori Sato)

kintone で検索クエリを実行する Python プログラム（Swagger版）

Last updated at Posted at 2025-05-09

下記の記事のプログラムに、ユーザーインターフェース（Python Flask、OpenAPI（Swagger））を追加しました。flasgger 用のコメントを自分で書くことは難しく、ChatGPT を活用しています。

Swagger は PUT で、kintone は GET という点がミソです。

fields （任意）を指定することで、表示するフィールドを制御できます。指定しなければ、すべてのフィールドが返されます。

APIキー版

pip install flask flasgger
102_get_records_query.py
from flask import Flask, jsonify, request
from flasgger import Swagger
import base64
import urllib.request
import json
import os
import sys
from pathlib import Path

app = Flask(__name__)
app.json.ensure_ascii = False
swagger = Swagger(app)


kintone_api_key = os.environ.get("KINTONE_API_KEY")
if not kintone_api_key:
    print("Error: KINTONE_API_KEY must be set in environment variables")
    sys.exit(1)


@app.route("/")
def index():
    return " Python の Web サーバーです！"


@app.route('/get_records_query', methods=['PUT'])
def get_records_query():
    """
    検索クエリを実行する
    ---
    tags:
      - Kintone
    summary: Kintone レコード検索API
    description: アプリIDとクエリを指定してレコードを取得します。
    consumes:
      - application/x-www-form-urlencoded
    parameters:
      - in: formData
        name: app
        type: integer
        required: true
        description: アプリ ID
        example: 123
      - in: formData
        name: query
        type: string
        required: true
        description: クエリ式
        example: レコード番号 <= 456 and 印刷 not in ("確認") order by レコード番号 asc limit 500
      - in: formData
        name: fields
        type: array
        required: false
        description: レスポンスに含めるフィールドの配列（1行ごとに1項目）
        items:
          type: string
        collectionFormat: multi
        example:
          - レコード番号
          - 印刷
          - 作成日時
    responses:
      200:
        description: 正常にレコードを取得
        schema:
          type: object
          properties:
            records:
              type: array
              items:
                type: object
            totalCount:
              type: integer
      400:
        description: 入力エラー
      500:
        description: サーバーエラー
        schema:
          type: object
          properties:
            error_code:
              type: integer
              example: 5000
            error_msg:
              type: string
              example: 認証に失敗しました
    """
    try:
        app_id = request.form.get("app", type=int)
        query = request.form.get("query", type=str)
        fields = request.form.getlist("fields")

        uri = "https://sample.cybozu.com/k/v1/records.json"

        binary_bytes = base64.b64decode(kintone_api_key)
        binary_string = binary_bytes.decode('latin1')

        headers = {
            "X-Cybozu-API-Token": binary_string,
            "Content-Type": "application/json",
        }

        body = {
            "app": app_id,
            "query": query,
            "fields": fields
        }

        req = urllib.request.Request(
                    url=uri,
                    data=json.dumps(body).encode(),
                    headers=headers,
                    method="GET",
                    )

        response = urllib.request.urlopen(req)
        res_dict = json.load(response)

        return jsonify(res_dict)

    except Exception as e:
        return jsonify({"error_code": 5000, "error_msg": str(e)}), 500

if __name__ == "__main__":
    app.run(debug=True, port=5000)
実行
# アプリIDに対する Kintone の API Key を設定します
export KINTONE_API_KEY="..."
python 102_get_records_query.py

　
画面表示
http://127.0.0.1:5000/apidocs/

　

　

　

「Try it out」ボタンを押下します。送信パラメータが編集可能になるので、アプリID（app）、クエリ（query）、表示するフィールド（fields、任意）を指定して、「Execute」ボタンを押下します。

　
他の事例ですが、結果は次のようになります。

　

ユーザー名／パスワード版

103_get_records_query.py
from flask import Flask, jsonify, request
from flasgger import Swagger
import base64
import urllib.request
import json
import os
import sys
from pathlib import Path

app = Flask(__name__)
app.json.ensure_ascii = False
swagger = Swagger(app)


kintone_username = os.environ.get("KINTONE_USERNAME")
if not kintone_username:
    print("Error: KINTONE_USERNAME must be set in environment variables")
    sys.exit(1)

kintone_password = os.environ.get("KINTONE_PASSWORD")
if not kintone_password:
    print("Error: KINTONE_PASSWORD must be set in environment variables")
    sys.exit(1)


@app.route("/")
def index():
    return " Python の Web サーバーです！"


@app.route('/get_records_query', methods=['PUT'])
def get_records_query():
    """
    検索クエリを実行する
    ---
    tags:
      - Kintone
    summary: Kintone レコード検索API
    description: アプリIDとクエリを指定してレコードを取得します。
    consumes:
      - application/x-www-form-urlencoded
    parameters:
      - in: formData
        name: app
        type: integer
        required: true
        description: アプリ ID
        example: 123
      - in: formData
        name: query
        type: string
        required: true
        description: クエリ式
        example: レコード番号 <= 456 and 印刷 not in ("確認") order by レコード番号 asc limit 500
      - in: formData
        name: fields
        type: array
        required: false
        description: レスポンスに含めるフィールドの配列（1行ごとに1項目）
        items:
          type: string
        collectionFormat: multi
        example:
          - レコード番号
          - 印刷
          - 作成日時
    responses:
      200:
        description: 正常にレコードを取得
        schema:
          type: object
          properties:
            records:
              type: array
              items:
                type: object
            totalCount:
              type: integer
      400:
        description: 入力エラー
      500:
        description: サーバーエラー
        schema:
          type: object
          properties:
            error_code:
              type: integer
              example: 5000
            error_msg:
              type: string
              example: 認証に失敗しました
    """
    try:
        app_id = request.form.get("app", type=int)
        query = request.form.get("query", type=str)
        fields = request.form.getlist("fields")

        uri = "https://sample.cybozu.com/k/v1/records.json"

        username_password = f"{kintone_username}:{kintone_password}"
        auth_header = base64.b64encode(username_password.encode())

        headers = {
            "Host": "sample.cybozu.com:443",
            "X-Cybozu-Authorization": auth_header,
            "Content-Type": "application/json",
        }

        body = {
            "app": app_id,
            "query": query,
            "fields": fields
        }

        req = urllib.request.Request(
                    url=uri,
                    data=json.dumps(body).encode(),
                    headers=headers,
                    method="GET",
                    )

        response = urllib.request.urlopen(req)
        res_dict = json.load(response)

        return jsonify(res_dict)

    except Exception as e:
        return jsonify({"error_code": 5000, "error_msg": str(e)}), 500

if __name__ == "__main__":
    app.run(debug=True, port=5000)
実行
# Kintone のユーザー名とパスワードを設定します
export KINTONE_USERNAME="..."
export KINTONE_PASSWORDE="..."
python 103_get_records_query.py

　
画面表示
http://127.0.0.1:5000/apidocs/

　
以上、

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?