Moderate Example

Moderate example for v3/tickets/query API - Complete pagination workflow.

This example demonstrates using OpenSearch filters and field selection to retrieve specific tickets across multiple pages using Point in Time (PIT) 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"

def get_access_token():
    """Authenticate and get JWT access 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"}
    )
    response.raise_for_status()
    return response.json()["access_token"]

def fetch_tickets_with_pagination(access_token):
    """Fetch tickets using the query API with PIT pagination.

    Returns all tickets matching the query by following pit_id/search_after pagination.
    """
    url = f"{BASE_URL}/v3/tickets/query"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "client_id": CLIENT_ID
    }

    # Calculate time range (last 30 days)
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=30)

    # Initial request payload with filters and field selection
    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": 1000,
        "includes_eu": False,
        "fields": [
            "ticket.number",
            "ticket.type",
            "ticket.state",
            "ticket.short_description",
            "ticket.resolved",
            "atr.domain_name"
        ],
        "filters": [
            {
                "term": {"ticket.resolved": True}
            }
        ]
    }

    all_tickets = []
    page = 1

    logger.info(f"Fetching tickets from {start_time} to {end_time}...")
    logger.info("Filter: resolved tickets only")

    while True:
        logger.info(f"Fetching page {page}...")

        response = requests.post(url, json=payload, headers=headers)
        response.raise_for_status()
        data = response.json()

        hits = data.get("data", {}).get("hits", [])
        all_tickets.extend(hits)
        logger.info(f"Retrieved {len(hits)} tickets (total: {len(all_tickets)})")

        # Check pagination
        pagination = data.get("meta", {}).get("pagination", {})
        pit_id = pagination.get("pit_id")
        search_after = pagination.get("search_after")

        if not pit_id or not search_after:
            logger.info("Last page reached.")
            break

        # Build pagination request
        payload = {
            "pit_id": pit_id,
            "search_after": search_after,
            "size": 1000,
            "fields": [
                "ticket.number",
                "ticket.type",
                "ticket.state",
                "ticket.short_description",
                "ticket.resolved",
                "atr.domain_name"
            ]
        }

        page += 1

    return all_tickets

def main():
    """Main execution."""
    logger.info("=" * 60)
    logger.info("v3/tickets/query API Example")
    logger.info("=" * 60)

    # Step 1: Get access token
    logger.info("1. Authenticating...")
    try:
        access_token = get_access_token()
        logger.info("Authentication successful")
    except Exception as e:
        logger.error(f"Authentication failed: {e}")
        return

    # Step 2: Fetch tickets with pagination
    logger.info("2. Fetching tickets...")
    try:
        tickets = fetch_tickets_with_pagination(access_token)
        logger.info(f"Successfully retrieved {len(tickets)} total tickets")
    except Exception as e:
        logger.error(f"Failed to fetch tickets: {e}")
        return

    # Step 3: Process results
    logger.info("3. Sample results:")
    for i, hit in enumerate(tickets[:3], 1):
        source = hit.get("_source", {})
        ticket = source.get("ticket", {})
        atr = source.get("atr", {})
        logger.info(f"Ticket {i}:")
        logger.info(f"  - Number: {ticket.get('number', 'N/A')}")
        logger.info(f"  - Type: {ticket.get('type', 'N/A')}")
        logger.info(f"  - State: {ticket.get('state', 'N/A')}")
        logger.info(f"  - Description: {ticket.get('short_description', 'N/A')}")
        logger.info(f"  - Domain: {atr.get('domain_name', 'N/A')}")

    if len(tickets) > 3:
        logger.info(f"... and {len(tickets) - 3} more tickets")

    logger.info("=" * 60)
    logger.info("Complete!")
    logger.info("=" * 60)

if __name__ == "__main__":
    main()