概要

本記事では、GitHub Copilot を使用して TROCCO REST API を Python で実行する手順を紹介します。Web API に対する理解が浅いと、API を実行するプログラムの開発で苦戦することがあります。そこで、AI の力を借りてこれらの作業を効率化できないかと考えました。本記事では以下の3つの手順を実施します。

GitHub Copilot Chat in GitHub を使用してクラスを記述
TROCCO 上でジョブの作成とトークンの取得
ノートブック型 Python 環境（Databricks）から TROCCO REST API を実行

最終的なコードはこちら

import time
import requests
from typing import Optional

class TroccoAPIClient:
    """Trocco APIクライアントクラス。

    Args:
        api_key (str): Trocco APIのAPIキー。
        wait_time (int, optional): リクエスト間の待機時間（秒）。デフォルトは30秒。
        max_wait_time (int, optional): 最大待機時間（秒）。デフォルトは3600秒（60分）。

    Attributes:
        api_key (str): Trocco APIのAPIキー。
        base_url (str): Trocco APIのベースURL。
        headers (dict): リクエストヘッダー。
        wait_time (int): リクエスト間の待機時間（秒）。
        max_wait_time (int): 最大待機時間（秒）。
    """

    def __init__(self, api_key: str, wait_time: int = 30, max_wait_time: int = 3600) -> None:
        self.api_key = api_key
        self.base_url = "https://trocco.io/api"
        self.headers = {
            "Authorization": f"Token {self.api_key}",
            "Content-Type": "application/json"
        }
        self.wait_time = wait_time
        self.max_wait_time = max_wait_time

    def post_job(self, payload: dict) -> Optional[dict]:
        """ジョブを作成する。

        Args:
            payload (dict): ジョブ作成に必要なデータ。

        Returns:
            Optional[dict]: APIからのレスポンス。失敗した場合はNone。
        """
        url = f"{self.base_url}/jobs"
        try:
            response = requests.post(url, json=payload, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"エラーが発生しました: {e}")
            return None

    def get_job(self, job_id: str) -> Optional[dict]:
        """特定のジョブの詳細を取得する。

        Args:
            job_id (str): 取得したいジョブのID。

        Returns:
            Optional[dict]: ジョブの詳細。失敗した場合はNone。
        """
        url = f"{self.base_url}/jobs/{job_id}"
        try:
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"エラーが発生しました: {e}")
            return None

    def get_jobs(self, params: dict = {}) -> Optional[dict]:
        """ジョブの一覧を取得する。

        Args:
            params (dict, optional): クエリパラメータ。

        Returns:
            Optional[dict]: ジョブの一覧。失敗した場合はNone。
        """
        url = f"{self.base_url}/jobs"
        try:
            response = requests.get(url, headers=self.headers, params=params)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"エラーが発生しました: {e}")
            return None

    def execute_job_and_wait(self, payload: dict) -> Optional[dict]:
        """ジョブを作成し、完了するまで待機して結果を取得する。

        Args:
            payload (dict): ジョブ作成に必要なデータ。

        Returns:
            Optional[dict]: 最終的なジョブの結果。失敗した場合はNone。
        """
        job = self.post_job(payload)
        if not job:
            return None

        job_id = job.get("id")
        if not job_id:
            print("ジョブIDを取得できませんでした。")
            return None

        start_time = time.time()
        while True:
            job_status = self.get_job(job_id)
            if job_status:
                status = job_status.get("status")
                if status == "succeeded":
                    print("ジョブが成功しました。")
                    return job_status
                elif status == "error":
                    print("ジョブが失敗しました。")
                    return job_status
                else:
                    print(f"ジョブのステータス: {status}。待機します。")
            else:
                print("ジョブステータスの取得に失敗しました。リトライします。")

            if time.time() - start_time > self.max_wait_time:
                print("最大待機時間を超えました。")
                return None

            time.sleep(self.wait_time)

# TroccoAPIClientのインスタンスを作成
api_key = "vV5vN6sCNivahnSYPYzzzzzzzzzzzz"  # ここにAPIキーを入力してください
client = TroccoAPIClient(api_key)

# ジョブ作成のためのペイロードを定義
payload = {
    "job_definition_id": "287317"
}

# ジョブを実行し、完了まで待機
result = client.execute_job_and_wait(payload)

# 結果を表示
if result:
    print("ジョブの結果:")
    print(result)
else:
    print("ジョブの取得に失敗しました。")

TROCCO REST API を利用するメリットとして、外部のワークフロー機能を活用できる点があります。TROCCO にもワークフロー機能はありますが、場合によっては外部のワークフロー機能から実行したいことがあります。本記事を参考に、ユースケースに応じた使い分けを実施し、TROCCO をより活用できるようになれば幸いです。

また、 GitHub Copilot Chat in GitHub 上でコードを生成しておりますが、 WEB 検索機能をもつ生成 AI サービスである ChatGPT や Microsoft Copilot でも同様にコードを生成できるはずです。出力されるコードを比較することで生成 AI サービスの特性を理解することができるかもしれません。

コードの作成と実行

1. GitHub Copilot Chat in GitHub を使用してクラスを記述

1-1. コードに含める API の特定

ジョブ実行に関連する以下の3つの API をコードに含めることにしました。

引用元：trocco API - 転送ジョブ実行

1-2. GitHub Copilot Chat in GitHub でコードを作成

以下の2つのプロンプトを使用して、コードを生成しました。

# これから提示するリンク先の URL の API を、制約を満たすような Python のクラスと実行コードとして記述してください。

## 制約

- 認証情報の設定、リクエストヘッダー、必要なパラメータなどを含める。
- Python の `requests` ライブラリを使用する。
- API エンドポイントにリクエストを送信し、レスポンスを表示する。
- Jupyter のノートブック環境から実行することを前提とする。
- 実行する API が非同期処理の場合は、実行と実行結果を取得するメソッドを追加すること。
- 待機時間（デフォルト30秒）と最大待機時間（デフォルト60分）をパラメータで指定できるようにする。
- 一時的なエラーが発生した場合に自動的にリトライすること。
- ベース URL は https://trocco.io/api とする。
- 認証方法は、`Authorization: Token {{API KEY}}` と指定する。
- Google Style の Docstring とデータ型ヒントを記述すること。

## リンク先

- https://documents.trocco.io/apidocs/post-job
- https://documents.trocco.io/apidocs/get-job
- https://documents.trocco.io/apidocs/get-jobs

1-3. 作成されたコードをコピー

コードはこちら

import time
import requests
from typing import Optional

class TroccoAPIClient:
    """Trocco APIクライアントクラス。

    Args:
        api_key (str): Trocco APIのAPIキー。
        wait_time (int, optional): リクエスト間の待機時間（秒）。デフォルトは30秒。
        max_wait_time (int, optional): 最大待機時間（秒）。デフォルトは3600秒（60分）。

    Attributes:
        api_key (str): Trocco APIのAPIキー。
        base_url (str): Trocco APIのベースURL。
        headers (dict): リクエストヘッダー。
        wait_time (int): リクエスト間の待機時間（秒）。
        max_wait_time (int): 最大待機時間（秒）。
    """

    def __init__(self, api_key: str, wait_time: int = 30, max_wait_time: int = 3600) -> None:
        self.api_key = api_key
        self.base_url = "https://api.trocco.io"
        self.headers = {
            "Authorization": f"Token {self.api_key}",
            "Content-Type": "application/json"
        }
        self.wait_time = wait_time
        self.max_wait_time = max_wait_time

    def post_job(self, payload: dict) -> Optional[dict]:
        """ジョブを作成する。

        Args:
            payload (dict): ジョブ作成に必要なデータ。

        Returns:
            Optional[dict]: APIからのレスポンス。失敗した場合はNone。
        """
        url = f"{self.base_url}/jobs"
        try:
            response = requests.post(url, json=payload, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"エラーが発生しました: {e}")
            return None

    def get_job(self, job_id: str) -> Optional[dict]:
        """特定のジョブの詳細を取得する。

        Args:
            job_id (str): 取得したいジョブのID。

        Returns:
            Optional[dict]: ジョブの詳細。失敗した場合はNone。
        """
        url = f"{self.base_url}/jobs/{job_id}"
        try:
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"エラーが発生しました: {e}")
            return None

    def get_jobs(self, params: dict = {}) -> Optional[dict]:
        """ジョブの一覧を取得する。

        Args:
            params (dict, optional): クエリパラメータ。

        Returns:
            Optional[dict]: ジョブの一覧。失敗した場合はNone。
        """
        url = f"{self.base_url}/jobs"
        try:
            response = requests.get(url, headers=self.headers, params=params)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"エラーが発生しました: {e}")
            return None

    def execute_job_and_wait(self, payload: dict) -> Optional[dict]:
        """ジョブを作成し、完了するまで待機して結果を取得する。

        Args:
            payload (dict): ジョブ作成に必要なデータ。

        Returns:
            Optional[dict]: 最終的なジョブの結果。失敗した場合はNone。
        """
        job = self.post_job(payload)
        if not job:
            return None

        job_id = job.get("id")
        if not job_id:
            print("ジョブIDを取得できませんでした。")
            return None

        start_time = time.time()
        while True:
            job_status = self.get_job(job_id)
            if job_status:
                status = job_status.get("status")
                if status == "success":
                    print("ジョブが成功しました。")
                    return job_status
                elif status == "failure":
                    print("ジョブが失敗しました。")
                    return job_status
                else:
                    print(f"ジョブのステータス: {status}。待機します。")
            else:
                print("ジョブステータスの取得に失敗しました。リトライします。")

            if time.time() - start_time > self.max_wait_time:
                print("最大待機時間を超えました。")
                return None

            time.sleep(self.wait_time)

# TroccoAPIClientのインスタンスを作成
api_key = "YOUR_API_KEY"  # ここにAPIキーを入力してください
client = TroccoAPIClient(api_key)

# ジョブ作成のためのペイロードを定義
payload = {
    # 必要なパラメータをここに記述します
    # 例:
    # "job_type": "your_job_type",
    # "config": { ... }
}

# ジョブを実行し、完了まで待機
result = client.execute_job_and_wait(payload)

# 結果を表示
if result:
    print("ジョブの結果:")
    print(result)
else:
    print("ジョブの取得に失敗しました。")

2. TROCCO 上でジョブの作成とトークンの取得

2-1. TROCCO で API で実行するジョブの作成と `job_definition_id` の取得

job_definition_id は、ジョブを TROCCO 上で表示した際の URL、またはジョブ名の横から取得できます。

2-2. TROCCO API KEY の取得

3. ノートブック型 Python 環境（Databricks）から TROCCO REST API を実行

3-1. コードを貼り付け

3-2. コードの検証と実行

api_key に TROCCO API KEY を、payload 変数に job_definition_id をキーとして取得済みの job_definition_id を設定し、実行します。

# TroccoAPIClientのインスタンスを作成
api_key = "vV5vN6sCNivahnSYPYzzzzzzzzzzzz"  # ここにAPIキーを入力してください
client = TroccoAPIClient(api_key)

# ジョブ作成のためのペイロードを定義
payload = {
    "job_definition_id": "287317"
}

# ジョブを実行し、完了まで待機
result = client.execute_job_and_wait(payload)

# 結果を表示
if result:
    print("ジョブの結果:")
    print(result)
else:
    print("ジョブの取得に失敗しました。")

ここでエラーもなく実行できればよかったのですが、以下の修正が必要でした。

URL が https://api.trocco.io となっていたため、https://trocco.io/api に修正
成功のステータスが success となっていたため、succeeded に修正 *1
エラーのステータスが failure となっていたため、error に修正 *1

*1 ジョブのステータスはドキュメントで以下のように記載されています。

引用元：trocco API - 転送ジョブ実行結果取得

修整後のコードはこちら

import time
import requests
from typing import Optional

class TroccoAPIClient:
    """Trocco APIクライアントクラス。

    Args:
        api_key (str): Trocco APIのAPIキー。
        wait_time (int, optional): リクエスト間の待機時間（秒）。デフォルトは30秒。
        max_wait_time (int, optional): 最大待機時間（秒）。デフォルトは3600秒（60分）。

    Attributes:
        api_key (str): Trocco APIのAPIキー。
        base_url (str): Trocco APIのベースURL。
        headers (dict): リクエストヘッダー。
        wait_time (int): リクエスト間の待機時間（秒）。
        max_wait_time (int): 最大待機時間（秒）。
    """

    def __init__(self, api_key: str, wait_time: int = 30, max_wait_time: int = 3600) -> None:
        self.api_key = api_key
        self.base_url = "https://trocco.io/api"
        self.headers = {
            "Authorization": f"Token {self.api_key}",
            "Content-Type": "application/json"
        }
        self.wait_time = wait_time
        self.max_wait_time = max_wait_time

    def post_job(self, payload: dict) -> Optional[dict]:
        """ジョブを作成する。

        Args:
            payload (dict): ジョブ作成に必要なデータ。

        Returns:
            Optional[dict]: APIからのレスポンス。失敗した場合はNone。
        """
        url = f"{self.base_url}/jobs"
        try:
            response = requests.post(url, json=payload, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"エラーが発生しました: {e}")
            return None

    def get_job(self, job_id: str) -> Optional[dict]:
        """特定のジョブの詳細を取得する。

        Args:
            job_id (str): 取得したいジョブのID。

        Returns:
            Optional[dict]: ジョブの詳細。失敗した場合はNone。
        """
        url = f"{self.base_url}/jobs/{job_id}"
        try:
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"エラーが発生しました: {e}")
            return None

    def get_jobs(self, params: dict = {}) -> Optional[dict]:
        """ジョブの一覧を取得する。

        Args:
            params (dict, optional): クエリパラメータ。

        Returns:
            Optional[dict]: ジョブの一覧。失敗した場合はNone。
        """
        url = f"{self.base_url}/jobs"
        try:
            response = requests.get(url, headers=self.headers, params=params)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"エラーが発生しました: {e}")
            return None

    def execute_job_and_wait(self, payload: dict) -> Optional[dict]:
        """ジョブを作成し、完了するまで待機して結果を取得する。

        Args:
            payload (dict): ジョブ作成に必要なデータ。

        Returns:
            Optional[dict]: 最終的なジョブの結果。失敗した場合はNone。
        """
        job = self.post_job(payload)
        if not job:
            return None

        job_id = job.get("id")
        if not job_id:
            print("ジョブIDを取得できませんでした。")
            return None

        start_time = time.time()
        while True:
            job_status = self.get_job(job_id)
            if job_status:
                status = job_status.get("status")
                if status == "succeeded":
                    print("ジョブが成功しました。")
                    return job_status
                elif status == "error":
                    print("ジョブが失敗しました。")
                    return job_status
                else:
                    print(f"ジョブのステータス: {status}。待機します。")
            else:
                print("ジョブステータスの取得に失敗しました。リトライします。")

            if time.time() - start_time > self.max_wait_time:
                print("最大待機時間を超えました。")
                return None

            time.sleep(self.wait_time)

3-3. 実行結果を確認

転送ジョブが成功したことを確認します。

まとめ

本記事では、TROCCO REST API を Python から実行する手順を紹介しました。AI が提示したコードが必ずしも正確ではないことを伝えるため、あえて手動で修正する手順を含めました。AI にエラーのないコードを生成してもらいたい場合、実装してもらう機能を減らすことで、期待通りに動作する可能性が高まります。GitHub Copilot を活用しつつ、必要に応じて手動での修正を加えることで、効率的に開発を進めることができます。AI ツールと適切に連携することで、開発効率を向上させる一助になれば幸いです。