🧙
TROCCO転送設定一覧取得APIを活用した効率的な棚卸しガイド

paulxll
2025/07/16に公開
TROCCO
tech
 概要TROCCO転送設定一覧APIが2025年7月にリリースされました。

転送設定一覧APIを使用して、大量の転送設定を効率的に棚卸しする方法を紹介します。

このAPIを活用することで、手動では難しい大規模な転送設定の管理や分析が可能になります。
APIの基本的な仕様についてはこちら:

https://documents.trocco.io/apidocs/get-job-definitions
以下に、棚卸しを始めるための事前準備および想定されるユースケース別の活用例をPythonのコードサンプルつきで記載しました。

Google Colab上で実行すればMac/Windowsどちらでも利用可能、環境構築不要で利用できますのでぜひお役立てください。

 事前準備Pythonの実行環境: Python 3.11以降を推奨し、以下のコードはGoogle Colabで動作確認しています(Google Colabは3.11でそのまま利用可能です）。
なお、以下のコードはGistで公開しています

https://gist.github.com/paulxll/71048610da646e708550897fafceae48
Google ColabでのAPIキー設定:
左サイドバーの鍵アイコンをクリックします。
「新しいシークレットを追加」で名前を TROCCO_API_KEY、値にAPIキーを設定します。
ノートブックアクセスを有効化します。
import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional

# Google Colab環境でのAPIキー取得（推奨）
try:
    from google.colab import userdata
    API_KEY = userdata.get("TROCCO_API_KEY")  # Colab Secretsから取得
except ImportError:
    # ローカル環境や他の環境での直接指定
    API_KEY = "your_api_key_here"

BASE_URL = "https://trocco.io"
HEADERS = {
    "Authorization": f"Token {API_KEY}",
    "Content-Type": "application/json"
}

def get_all_job_definitions() -> List[Dict]:
    """
    すべての転送設定を取得する（ページネーション対応）
    """
    all_items = []
    cursor = None
    
    while True:
        params = {"limit": 100}
        if cursor:
            params["cursor"] = cursor
            
        response = requests.get(
            f"{BASE_URL}/api/job_definitions",
            headers=HEADERS,
            params=params
        )
        response.raise_for_status()
        
        data = response.json()
        all_items.extend(data["items"])
        
        cursor = data.get("next_cursor")
        if not cursor:
            break
            
    return all_items

 活用例
 1. 特定の名称を含む転送設定を探す活用シーン:
プロジェクト終了時の設定棚卸し
命名規則チェック
環境別設定の確認
APIのnameパラメータは、部分一致検索に加えて正規表現にも対応しています。正規表現を使用する場合は、パターンを/で囲んで指定します。
def find_by_name_pattern(pattern: str, use_regex: bool = False) -> List[Dict]:
    """
    名称パターンで転送設定を検索（正規表現対応、ページネーション自動処理）
    """
    if use_regex:
        # 正規表現パターンは / で囲む
        search_pattern = f"/{pattern}/"
    else:
        # 通常の部分一致検索
        search_pattern = pattern
    
    all_items = []
    cursor = None
    
    try:
        while True:
            params = {"name": search_pattern, "limit": 100}
            if cursor:
                params["cursor"] = cursor
                
            response = requests.get(
                f"{BASE_URL}/api/job_definitions",
                headers=HEADERS,
                params=params
            )
            response.raise_for_status()
            
            data = response.json()
            all_items.extend(data["items"])
            
            cursor = data.get("next_cursor")
            if not cursor:
                break
                
        return all_items
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 400:
            print(f"検索パターンエラー: {search_pattern}")
            return []
        raise

# 基本的な部分一致検索
legacy_jobs = find_by_name_pattern("legacy")
test_jobs = find_by_name_pattern("test")

# 正規表現を使った高度な検索
def find_environment_settings():
    """
    環境別(prod, stg, dev)の名称をつけられている設定の棚卸し
    """
    # prod, stg, dev のいずれかを含む設定を検索
    env_jobs = find_by_name_pattern("prod|stg|dev", use_regex=True)
    
    # 環境別に分類
    env_classification = {"prod": [], "stg": [], "dev": [], "other": []}
    
    for job in env_jobs:
        name = job["name"].lower()
        if "prod" in name:
            env_classification["prod"].append(job)
        elif "stg" in name or "staging" in name:
            env_classification["stg"].append(job)
        elif "dev" in name or "development" in name:
            env_classification["dev"].append(job)
        else:
            env_classification["other"].append(job)
    
    return env_classification

# 使用例
print(f"legacy含む設定: {len(legacy_jobs)}件")
for job in legacy_jobs:
    job_name = job["name"]
    job_id = job["id"]
    print(f"  - {job_name} (ID: {job_id})")

# 環境別設定の確認
env_settings = find_environment_settings()
print("\n【環境別設定の分布】")
for env, jobs in env_settings.items():
    print(f"{env.upper()}環境: {len(jobs)}件")
    if len(jobs) > 0:
        for job in jobs[:3]:  # 最初の3件のみ表示
            job_name = job["name"]
            print(f"  - {job_name}")
        if len(jobs) > 3:
            remaining_count = len(jobs) - 3
            print(f"  ... 他{remaining_count}件")

# その他の正規表現パターン例
def advanced_regex_examples():
    """
    正規表現の活用例
    """
    patterns = {
        "数字で終わる": r"\d+$",
        "test_で始まる": r"^test_",
        "daily_hourly含む": r"daily|hourly", 
        "括弧付き": r"\([^)]+\)",
        "バックアップ系": r"backup|archive|copy"
    }
    
    results = {}
    for desc, pattern in patterns.items():
        try:
            jobs = find_by_name_pattern(pattern, use_regex=True)
            results[desc] = len(jobs)
        except Exception as e:
            results[desc] = f"エラー: {e}"
    
    return results

# 正規表現パターンの実行例
pattern_results = advanced_regex_examples()
print("\n【正規表現パターン検索結果】")
for pattern, count in pattern_results.items():
    print(f"{pattern}: {count}")

 2. 特定の人が作成した転送設定を探す活用シーン:
退職者の設定引き継ぎ
作成者別の設定管理
責任者変更時の設定確認
def find_by_creator(creator_email: str) -> List[Dict]:
    """
    作成者で転送設定をフィルタリング
    """
    all_jobs = get_all_job_definitions()
    return [
        job for job in all_jobs 
        if job.get("created_by") and job["created_by"].get("email") == creator_email
    ]

def find_by_multiple_creators(creator_emails: List[str]) -> Dict[str, List[Dict]]:
    """
    複数の作成者の設定を一括取得
    """
    all_jobs = get_all_job_definitions()
    result = {}
    
    for email in creator_emails:
        result[email] = [
            job for job in all_jobs 
            if job.get("created_by") and job["created_by"].get("email") == email
        ]
    
    return result

# 使用例：退職者の設定を特定
departed_employees = ["wajima@exapmle.com", "nakajima@exapmle.com"]
departed_jobs = find_by_multiple_creators(departed_employees)

for email, jobs in departed_jobs.items():
    print(f"{email}が作成した設定: {len(jobs)}件")
    for job in jobs:
        job_name = job["name"]
        update_date = job["updated_at"][:10]
        print(f"  - {job_name} (更新日: {update_date})")

 3. 最新更新された/されていない転送設定を探す活用シーン:
使われていない設定の特定
定期メンテナンスの優先度判定
古い設定の見直し
def find_by_update_period(days_ago: int, recently_updated: bool = True) -> List[Dict]:
    """
    更新日時で転送設定をフィルタリング
    """
    all_jobs = get_all_job_definitions()
    # 比較のためにカットオフ日時をタイムゾーン対応にする
    cutoff_date = datetime.now().astimezone() - timedelta(days=days_ago)


    filtered_jobs = []
    for job in all_jobs:
        # パースした日時もタイムゾーン対応であることを確認
        updated_at = datetime.fromisoformat(job["updated_at"].replace("Z", "+00:00"))

        if recently_updated:
            # 最近更新された設定
            if updated_at >= cutoff_date:
                filtered_jobs.append(job)
        else:
            # 長期間更新されていない設定
            if updated_at < cutoff_date:
                filtered_jobs.append(job)

    # 更新日時でソート
    return sorted(filtered_jobs, key=lambda x: x["updated_at"], reverse=recently_updated)

# 使用例
recently_updated = find_by_update_period(30, recently_updated=True)
old_jobs = find_by_update_period(365, recently_updated=False)

print(f"過去30日に更新された設定: {len(recently_updated)}件")
print(f"1年以上更新されていない設定: {len(old_jobs)}件")

# 古い設定の詳細表示
print("\n【注意が必要な古い設定】")
for job in old_jobs[:10]:  # 上位10件
    days_old = (datetime.now().astimezone() - datetime.fromisoformat(job["updated_at"].replace("Z", "+00:00"))).days
    job_name = job["name"]
    created_by = job.get("created_by", {})
    creator_email = created_by.get("email", "N/A")
    print(f"  - {job_name} ({days_old}日前更新, 作成者: {creator_email})")

 4. 特定のリソースグループを含む・含まない転送設定を探す活用シーン:
チーム体制変更時の設定整理
権限管理の見直し
設定の所有者明確化
def find_by_resource_group(resource_group_id: Optional[int] = None, include: bool = True) -> List[Dict]:
    """
    リソースグループで転送設定をフィルタリング
    """
    all_jobs = get_all_job_definitions()
    
    if include:
        if resource_group_id is None:
            # リソースグループに所属している設定
            return [job for job in all_jobs if job.get("resource_group_id") is not None]
        else:
            # 特定のリソースグループに所属している設定
            return [job for job in all_jobs if job.get("resource_group_id") == resource_group_id]
    else:
        if resource_group_id is None:
            # リソースグループに所属していない設定
            return [job for job in all_jobs if job.get("resource_group_id") is None]
        else:
            # 特定のリソースグループに所属していない設定
            return [job for job in all_jobs if job.get("resource_group_id") != resource_group_id]

def analyze_resource_group_distribution() -> Dict[str, int]:
    """
    リソースグループ別の設定数を分析
    """
    all_jobs = get_all_job_definitions()
    distribution = {}
    
    for job in all_jobs:
        rg_id = job.get("resource_group_id")
        key = f"ResourceGroupId: {rg_id}" if rg_id else "未設定"
        distribution[key] = distribution.get(key, 0) + 1
    
    return distribution

# 使用例
unassigned_jobs = find_by_resource_group(include=False)
team_a_jobs = find_by_resource_group(resource_group_id=123, include=True)

print(f"リソースグループ未設定の設定: {len(unassigned_jobs)}件")
print(f"チームAの設定: {len(team_a_jobs)}件")

# リソースグループ分布の確認
distribution = analyze_resource_group_distribution()
print("\nリソースグループ別分布:")
for group, count in sorted(distribution.items(), key=lambda x: x[1], reverse=True):
    print(f"  {group}: {count}件")

 5. 特定のラベルを含む・含まない転送設定を探す活用シーン:
環境別設定の管理
用途別設定の棚卸し
ラベル付与の標準化
def find_by_labels(required_labels: List[str] = None, excluded_labels: List[str] = None) -> List[Dict]:
    """
    ラベルで転送設定をフィルタリング
    """
    all_jobs = get_all_job_definitions()
    filtered_jobs = []
    
    for job in all_jobs:
        job_labels = job.get("labels", []) or []
        
        # 必須ラベルのチェック
        if required_labels:
            if not all(label in job_labels for label in required_labels):
                continue
        
        # 除外ラベルのチェック
        if excluded_labels:
            if any(label in job_labels for label in excluded_labels):
                continue
        
        filtered_jobs.append(job)
    
    return filtered_jobs

def analyze_label_usage() -> Dict[str, int]:
    """
    ラベル使用状況を分析
    """
    all_jobs = get_all_job_definitions()
    label_count = {}
    
    for job in all_jobs:
        labels = job.get("labels", []) or []
        for label in labels:
            label_count[label] = label_count.get(label, 0) + 1
    
    return label_count

# 使用例
prod_jobs = find_by_labels(required_labels=["production"])
staging_without_test = find_by_labels(
    required_labels=["staging"], 
    excluded_labels=["test"]
)
unlabeled_jobs = find_by_labels(required_labels=[])

print(f"productionラベル付き設定: {len(prod_jobs)}件")
print(f"stagingラベル付き(testラベル除く)設定: {len(staging_without_test)}件")

# ラベル使用状況の分析
label_usage = analyze_label_usage()
print("\nラベル使用ランキング:")
for label, count in sorted(label_usage.items(), key=lambda x: x[1], reverse=True)[:10]:
    print(f"  {label}: {count}件")

# ラベル未設定の設定を特定
unlabeled_count = len([job for job in get_all_job_definitions() if not job.get("labels")])
print(f"\nラベル未設定の設定: {unlabeled_count}件")

 6. スケジュール設定の数や重複を探す活用シーン:
TROCCOアカウントに設定されている同時実行数上限を超えたスケジュール設定の検知
同時実行数上限による実行待機・遅延を防ぐための最適化
設定の複雑化チェック
重複実行の排除による運用負荷軽減
同一時間帯での実行集中の回避
def find_many_schedules_jobs(threshold: int = 10) -> List[Dict]:
    """
    スケジュール設定数が多い転送設定を特定
    """
    all_jobs = get_all_job_definitions()
    many_schedules_jobs = []
    
    for job in all_jobs:
        schedules = job.get("schedules", []) or []
        if len(schedules) >= threshold:
            many_schedules_jobs.append({
                "job_id": job["id"],
                "job_name": job["name"],
                "schedule_count": len(schedules),
                "schedules": schedules
            })
    
    return sorted(many_schedules_jobs, key=lambda x: x["schedule_count"], reverse=True)

def create_schedule_key(schedule: Dict) -> str:
    """
    スケジュール設定から一意のキーを生成（重複判定用）
    """
    # 必要な要素を組み合わせて一意のキーを作成
    key_elements = [
        schedule.get("frequency", ""),
        str(schedule.get("minute", "")),
        str(schedule.get("hour", "")),
        str(schedule.get("day", "")),
        str(schedule.get("day_of_week", "")),
        schedule.get("time_zone", "")
    ]
    return "|".join(key_elements)

def find_duplicate_schedules() -> Dict[str, List[Dict]]:
    """
    同一設定のスケジュールが設定されている転送設定を特定
    """
    all_jobs = get_all_job_definitions()
    schedule_to_jobs = {}
    
    for job in all_jobs:
        schedules = job.get("schedules", []) or []
        for schedule in schedules:
            schedule_key = create_schedule_key(schedule)
            
            if schedule_key not in schedule_to_jobs:
                schedule_to_jobs[schedule_key] = []
            
            schedule_to_jobs[schedule_key].append({
                "job_id": job["id"],
                "job_name": job["name"],
                "schedule": schedule
            })
    
    # 重複があるもの（2つ以上のジョブで同じスケジュール設定）のみを返す
    duplicates = {
        key: jobs for key, jobs in schedule_to_jobs.items() 
        if len(jobs) > 1
    }
    
    return duplicates

def analyze_schedule_duplicates() -> Dict:
    """
    重複スケジュール設定の詳細分析
    """
    duplicates = find_duplicate_schedules()
    
    analysis = {
        "duplicate_count": len(duplicates),
        "affected_jobs": set(),
        "schedule_details": []
    }
    
    for schedule_key, jobs in duplicates.items():
        # 同じスケジュール設定を持つジョブを記録
        for job in jobs:
            analysis["affected_jobs"].add(job["job_name"])
        
        # スケジュール設定の詳細を記録
        schedule_info = jobs[0]["schedule"]  # 最初のジョブのスケジュール設定
        analysis["schedule_details"].append({
            "schedule_key": schedule_key,
            "schedule_config": schedule_info,
            "job_count": len(jobs),
            "jobs": [job["job_name"] for job in jobs]
        })
    
    analysis["affected_jobs"] = list(analysis["affected_jobs"])
    return analysis

def find_unscheduled_jobs() -> List[Dict]:
    """
    スケジュール未設定の設定を特定
    """
    all_jobs = get_all_job_definitions()
    return [
        job for job in all_jobs 
        if not job.get("schedules") or len(job.get("schedules", [])) == 0
    ]

# 使用例
many_schedules_jobs = find_many_schedules_jobs(threshold=10)
duplicate_analysis = analyze_schedule_duplicates()
unscheduled_jobs = find_unscheduled_jobs()

print(f"スケジュール設定が10個以上の転送設定: {len(many_schedules_jobs)}件")
print("\n【スケジュール設定数の多い設定】")
for job in many_schedules_jobs[:5]:  # 上位5件
    job_name = job["job_name"]
    schedule_count = job["schedule_count"]
    print(f"  - {job_name} (スケジュール数: {schedule_count})")

duplicate_count = duplicate_analysis["duplicate_count"]
affected_jobs_count = len(duplicate_analysis["affected_jobs"])
print(f"\n重複スケジュール設定: {duplicate_count}種類")
print(f"影響を受ける転送設定: {affected_jobs_count}件")

print("\n【重複スケジュール設定の詳細】")
for detail in duplicate_analysis["schedule_details"][:5]:  # 上位5件
    schedule = detail["schedule_config"]
    frequency = schedule.get("frequency", "不明")
    time_zone = schedule.get("time_zone", "不明")
    hour = schedule.get("hour", "*")
    minute = schedule.get("minute", "0")
    job_count = detail["job_count"]
    jobs = detail["jobs"]
    jobs_display = jobs[:3]
    ellipsis = "..." if len(jobs) > 3 else ""
    print(f"  - {frequency}実行 ({time_zone})")
    print(f"    設定: {hour}:{minute}分")
    print(f"    対象設定: {job_count}件 - {jobs_display}{ellipsis}")
    print()

print(f"スケジュール未設定の設定: {len(unscheduled_jobs)}件")

 補足事項
 APIの利用制限について1回のリクエストで最大100件の転送設定を取得できます。
100件を超える転送設定がある場合は、ページネーション（cursor）を使用して全件取得できます。

APIコール制限: プランによって制限が異なります（詳細はAPIコール制限についてを参照してください）。
有償のプランをご利用の場合、この用途（棚卸し作業）では制限に達する可能性は低いですが、大量のリクエストを行いたい場合は適切な間隔でリクエストを実行してください。
このガイドを活用して、転送設定の効率的な管理と最適化を実現してください。
株式会社primeNumberPublication
「あらゆるデータを、ビジネスの力に変える」データテクノロジーカンパニー。クラウドETL「TROCCO」やクラウドデータカタログ「COMETA」、データテクノロジー領域の課題解決を実現するプロフェッショナルサービスを提供しています。
概要

事前準備

活用例

1. 特定の名称を含む転送設定を探す

2. 特定の人が作成した転送設定を探す

3. 最新更新された/されていない転送設定を探す

4. 特定のリソースグループを含む・含まない転送設定を探す

5. 特定のラベルを含む・含まない転送設定を探す

6. スケジュール設定の数や重複を探す

補足事項

APIの利用制限について

Discussion
