Simple Example

Simple example for v3/jobs/query API - Single request without pagination.

Use this when you want to retrieve a small number of job execution records without handling pagination.

---

Code Example

import requests
import logging
from datetime import datetime, timedelta

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Configuration
BASE_URL = "https://your-api-base-url.com"
TOKEN_URL = "https://your-auth-endpoint.com/oauth2/token"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"

# Step 1: Get access token
logger.info("Getting access token...")
token_response = requests.post(
    TOKEN_URL,
    data={
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    },
    headers={"Content-Type": "application/x-www-form-urlencoded"}
)
access_token = token_response.json()["access_token"]
logger.info("Authenticated")

# Step 2: Prepare request
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=7)

payload = {
    "application": "atr",
    "app_type": ["ec2", "kubernetes"],
    "domain": ["*"],
    "start_time": start_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
    "end_time": end_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
    "size": 500,
    "includes_eu": False
}

headers = {
    "Authorization": f"Bearer {access_token}",
    "Content-Type": "application/json",
    "client_id": CLIENT_ID
}

# Step 3: Make request
logger.info("Fetching jobs...")
response = requests.post(
    f"{BASE_URL}/v3/jobs/query",
    json=payload,
    headers=headers
)

# Step 4: Process response
if response.status_code == 200:
    data = response.json()
    hits = data.get("data", {}).get("hits", [])
    total = data.get("data", {}).get("total", 0)
    pit_id = data.get("meta", {}).get("pagination", {}).get("pit_id")

    logger.info(f"Retrieved {len(hits)} jobs (total matching: {total})")

    if pit_id:
        logger.info("More jobs available. Use pit_id and search_after to fetch next page.")

    # Display first job
    if hits:
        logger.info("First job:")
        source = hits[0].get("_source", {})
        job = source.get("job", {})
        workflow = source.get("workflow", {})
        atr = source.get("atr", {})
        logger.info(f"  Name: {job.get('name', 'N/A')}")
        logger.info(f"  State: {job.get('state', 'N/A')}")
        logger.info(f"  Workflow: {workflow.get('name', 'N/A')}")
        logger.info(f"  Domain: {atr.get('domain_name', 'N/A')}")
else:
    logger.error(f"Request failed: {response.status_code}")
    logger.error(f"  {response.text}")