TROCCO REST APIをPythonで操作：GitHub Copilot で試してみた結果

## 概要

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

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


<details>
  <summary>最終的なコードはこちら</summary>

```python
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)
```

```python
# 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("ジョブの取得に失敗しました。")
```

</details>

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 をコードに含めることにしました。

1. [trocco API - 転送ジョブ実行](https://documents.trocco.io/apidocs/post-job)
2. [trocco API - 転送ジョブ実行結果取得](https://documents.trocco.io/apidocs/get-job)
3. [trocco API - 転送ジョブ一覧取得](https://documents.trocco.io/apidocs/get-jobs)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/47790bf2-8b0d-1603-479f-382daf3b3170.png)

*引用元：[trocco API - 転送ジョブ実行](https://documents.trocco.io/apidocs/post-job)*

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

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

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

## 制約

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

```text
## リンク先

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9cca4655-a074-37af-5991-e62f9e5af930.png)

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


<details>
  <summary>コードはこちら</summary>

```python
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)
```

```python
# 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("ジョブの取得に失敗しました。")
```

</details>

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

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f5491bb2-5799-e222-57f5-00e42830512e.png)

#### 2-2. TROCCO API KEY の取得

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/09c599c1-dbfd-7e15-4c0f-755d0fa21632.png)

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/397abd30-8ec3-efd5-30ee-2659761ac4f6.png)

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

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

```python
# 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("ジョブの取得に失敗しました。")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3c0048b3-a20c-c8b4-3ea8-7b3e1ba328e9.png)

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

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6bc5f48f-001c-f334-f723-b377904abda8.png)

*引用元：[trocco API - 転送ジョブ実行結果取得](https://documents.trocco.io/apidocs/get-job)*


<details>
  <summary>修整後のコードはこちら</summary>

```python
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)
```

</details>

#### 3-3. 実行結果を確認

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b6ef03d5-f35b-14ea-f94a-8e8a9575dced.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b7f638cf-b09c-bc64-6ee8-083150e7873d.png)

## まとめ


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