"""Search operations for JIRA."""
import logging
from typing import Any, Dict, List, Optional
from jira.exceptions import JIRAError
from mcp_jira.client import JiraClient
logger = logging.getLogger(__name__)
class SearchOperations:
"""Handles JIRA search operations."""
def __init__(self, client: JiraClient):
"""Initialize search operations.
Args:
client: JiraClient instance
"""
self.client = client
def search_issues(
self,
jql: str,
start_at: int = 0,
max_results: int = 50,
fields: Optional[List[str]] = None,
expand: Optional[str] = None,
) -> Dict[str, Any]:
"""Search for issues using JQL.
Args:
jql: JQL query string
start_at: Starting index for pagination
max_results: Maximum number of results to return
fields: List of fields to retrieve (optional)
expand: Comma-separated list of entities to expand (optional)
Returns:
Dictionary with search results
Raises:
JIRAError: If search fails
"""
try:
jira = self.client.jira
logger.info(f"Searching issues with JQL: {jql}")
results = jira.search_issues(
jql,
startAt=start_at,
maxResults=max_results,
fields=fields,
expand=expand,
)
issues = []
for issue in results:
fields_dict = issue.raw.get("fields", {})
custom = self.client.custom_fields.extract_custom_fields(fields_dict)
issues.append({
"key": issue.key,
"id": issue.id,
"summary": fields_dict.get("summary"),
"status": fields_dict.get("status", {}).get("name"),
"assignee": fields_dict.get("assignee", {}).get("displayName") if fields_dict.get("assignee") else None,
"created": fields_dict.get("created"),
"updated": fields_dict.get("updated"),
"custom_fields": custom,
"url": f"{self.client.config.jira_url}/browse/{issue.key}",
})
return {
"total": results.total,
"start_at": start_at,
"max_results": max_results,
"issues": issues,
}
except JIRAError as e:
logger.error(f"Search failed: {e}")
raise
def get_all_issues(
self,
jql: str,
fields: Optional[List[str]] = None,
batch_size: int = 100,
) -> List[Dict[str, Any]]:
"""Get all issues matching JQL query (handles pagination automatically).
Args:
jql: JQL query string
fields: List of fields to retrieve (optional)
batch_size: Number of issues to fetch per request
Returns:
List of all matching issues
Raises:
JIRAError: If search fails
"""
all_issues = []
start_at = 0
while True:
result = self.search_issues(
jql=jql,
start_at=start_at,
max_results=batch_size,
fields=fields,
)
all_issues.extend(result["issues"])
if start_at + batch_size >= result["total"]:
break
start_at += batch_size
logger.info(f"Retrieved {len(all_issues)} total issues")
return all_issues