Advanced Example
Advanced examples for v3/tickets/query API showing different use cases.
Demonstrates:
- Basic ticket query
- Term filter for resolved tickets
- Field selection to reduce response size
- PIT pagination handling
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__)
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():
"""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"}
)
return response.json()["access_token"]
def get_headers(access_token):
"""Build request headers."""
return {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
"client_id": CLIENT_ID
}
def example_1_basic_query():
"""Example 1: Basic query for all tickets in a domain."""
logger.info("=" * 60)
logger.info("Example 1: Basic Query - All tickets from a specific domain")
logger.info("=" * 60)
access_token = get_access_token()
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=7)
payload = {
"application": "atr",
"app_type": ["ec2"],
"domain": ["your-domain.example.com"],
"start_time": start_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
"end_time": end_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
"size": 100,
"includes_eu": False
}
response = requests.post(
f"{BASE_URL}/v3/tickets/query",
json=payload,
headers=get_headers(access_token)
)
data = response.json()
logger.info(f"Status: {response.status_code}")
logger.info(f"Tickets retrieved: {data.get('data', {}).get('count', 0)}")
logger.info(f"Total matching: {data.get('data', {}).get('total', 0)}")
def example_2_resolved_tickets_filter():
"""Example 2: Query with term filter for resolved tickets of a specific type."""
logger.info("=" * 60)
logger.info("Example 2: Term Filters - Resolved SR_ITEM tickets")
logger.info("=" * 60)
access_token = get_access_token()
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=30)
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": 100,
"includes_eu": False,
"filters": [
{"term": {"ticket.resolved": True}},
{"term": {"ticket.type": "SR_ITEM"}}
]
}
response = requests.post(
f"{BASE_URL}/v3/tickets/query",
json=payload,
headers=get_headers(access_token)
)
data = response.json()
logger.info(f"Status: {response.status_code}")
logger.info(f"Filtered tickets found: {data.get('data', {}).get('total', 0)}")
def example_3_field_selection():
"""Example 3: Query with specific field selection to reduce response size."""
logger.info("=" * 60)
logger.info("Example 3: Field Selection - Only essential fields")
logger.info("=" * 60)
access_token = get_access_token()
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": 50,
"includes_eu": False,
"fields": [
"ticket.number",
"ticket.state",
"ticket.resolved"
]
}
response = requests.post(
f"{BASE_URL}/v3/tickets/query",
json=payload,
headers=get_headers(access_token)
)
data = response.json()
hits = data.get("data", {}).get("hits", [])
logger.info(f"Status: {response.status_code}")
logger.info(f"Tickets retrieved: {len(hits)}")
if hits:
logger.info("First ticket (only selected fields):")
logger.info(f" Fields returned: {list(hits[0].get('_source', {}).keys())}")
def example_4_pagination():
"""Example 4: Using pit_id and search_after for pagination."""
logger.info("=" * 60)
logger.info("Example 4: Pagination - Using pit_id and search_after")
logger.info("=" * 60)
access_token = get_access_token()
headers = get_headers(access_token)
url = f"{BASE_URL}/v3/tickets/query"
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=7)
# Initial request
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.state"]
}
all_tickets = []
page = 1
max_pages = 5
while True:
logger.info(f"Fetching page {page}...")
response = requests.post(url, json=payload, headers=headers)
data = response.json()
hits = data.get("data", {}).get("hits", [])
all_tickets.extend(hits)
logger.info(f"Page {page}: {len(hits)} tickets (total so far: {len(all_tickets)})")
logger.info(f"Rate limit remaining: {data.get('meta', {}).get('rate_limit', {}).get('daily_remaining', 'N/A')}")
# Check if last page
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
if page >= max_pages:
logger.info(f"Max pages ({max_pages}) reached, stopping.")
break
# Build pagination request
payload = {
"pit_id": pit_id,
"search_after": search_after,
"size": 1000,
"fields": ["ticket.number", "ticket.state"]
}
page += 1
logger.info(f"Total tickets retrieved: {len(all_tickets)}")
def main():
"""Run all examples."""
logger.info("=" * 60)
logger.info("v3/tickets/query API - Advanced Examples")
logger.info("=" * 60)
try:
example_1_basic_query()
logger.info("")
example_2_resolved_tickets_filter()
logger.info("")
example_3_field_selection()
logger.info("")
example_4_pagination()
logger.info("=" * 60)
logger.info("All examples completed!")
logger.info("=" * 60)
except Exception as e:
logger.error(f"Error: {e}")
if __name__ == "__main__":
main()