"""Custom fields handling for JIRA."""
import logging
from datetime import datetime
from functools import lru_cache
from typing import Any, Dict, List, Optional
from jira import JIRA
from mcp_jira.models.types import CustomField
logger = logging.getLogger(__name__)
class CustomFieldsManager:
"""Manages custom field resolution and type conversion."""
def __init__(self, jira_client: JIRA):
"""Initialize custom fields manager.
Args:
jira_client: Authenticated JIRA client instance
"""
self.jira_client = jira_client
self._field_cache: Optional[Dict[str, CustomField]] = None
self._name_to_id_map: Optional[Dict[str, str]] = None
def _load_fields(self) -> None:
"""Load all fields from JIRA and cache them."""
if self._field_cache is not None:
return
logger.info("Loading custom fields from JIRA")
fields = self.jira_client.fields()
self._field_cache = {}
self._name_to_id_map = {}
for field in fields:
field_id = field["id"]
field_name = field["name"]
is_custom = field.get("custom", False)
custom_field = CustomField(
id=field_id,
name=field_name,
custom=is_custom,
schema=field.get("schema"),
searchable=field.get("searchable", True),
)
self._field_cache[field_id] = custom_field
self._name_to_id_map[field_name.lower()] = field_id
logger.info(f"Loaded {len(self._field_cache)} fields from JIRA")
def get_field_by_id(self, field_id: str) -> Optional[CustomField]:
"""Get field metadata by ID.
Args:
field_id: Field ID (e.g., 'customfield_10001')
Returns:
CustomField metadata or None if not found
"""
self._load_fields()
return self._field_cache.get(field_id) if self._field_cache else None
def get_field_by_name(self, field_name: str) -> Optional[CustomField]:
"""Get field metadata by name (case-insensitive).
Args:
field_name: Field name
Returns:
CustomField metadata or None if not found
"""
self._load_fields()
if not self._name_to_id_map:
return None
field_id = self._name_to_id_map.get(field_name.lower())
if field_id and self._field_cache:
return self._field_cache.get(field_id)
return None
def get_field_id(self, field_name: str) -> Optional[str]:
"""Get field ID from field name.
Args:
field_name: Field name
Returns:
Field ID or None if not found
"""
field = self.get_field_by_name(field_name)
return field.id if field else None
def get_all_custom_fields(self) -> List[CustomField]:
"""Get all custom fields.
Returns:
List of custom fields
"""
self._load_fields()
if not self._field_cache:
return []
return [f for f in self._field_cache.values() if f.custom]
def convert_field_value(
self, field: CustomField, value: Any
) -> Any:
"""Convert field value to appropriate type based on field schema.
Args:
field: Field metadata
value: Raw value to convert
Returns:
Converted value appropriate for JIRA API
"""
if value is None:
return None
if not field.schema:
return value
field_type = field.schema.get("type", "")
custom_type = field.schema.get("custom", "")
# User picker fields
if field_type == "user" or "user" in custom_type.lower():
if isinstance(value, dict):
return value
# Assume it's an account ID or username
return {"accountId": value} if value.startswith("account") else {"name": value}
# Date fields
if field_type == "date" or custom_type.endswith(":date"):
if isinstance(value, datetime):
return value.strftime("%Y-%m-%d")
return value
# DateTime fields
if field_type == "datetime" or custom_type.endswith(":datetime"):
if isinstance(value, datetime):
return value.isoformat()
return value
# Array/multi-value fields
if field_type == "array":
if not isinstance(value, list):
value = [value]
items_type = field.schema.get("items", "")
if items_type == "string":
return [str(v) for v in value]
elif items_type == "option":
# Select list values
return [{"value": str(v)} if isinstance(v, str) else v for v in value]
return value
# Option/select fields
if field_type == "option" or custom_type.endswith(":select"):
if isinstance(value, dict):
return value
return {"value": str(value)}
# Number fields
if field_type == "number" or custom_type.endswith(":float"):
return float(value)
# String fields (default)
return value
def prepare_custom_fields(
self, custom_fields: Dict[str, Any]
) -> Dict[str, Any]:
"""Prepare custom fields dictionary for JIRA API.
Converts field names to IDs and values to appropriate types.
Args:
custom_fields: Dictionary with field names or IDs as keys
Returns:
Dictionary with field IDs as keys and converted values
"""
result = {}
for key, value in custom_fields.items():
# Check if key is already a field ID
if key.startswith("customfield_"):
field = self.get_field_by_id(key)
field_id = key
else:
# Try to resolve by name
field = self.get_field_by_name(key)
field_id = field.id if field else None
if not field_id:
logger.warning(f"Could not resolve custom field: {key}")
continue
if field:
converted_value = self.convert_field_value(field, value)
result[field_id] = converted_value
else:
# Field not found, use value as-is
result[field_id] = value
return result
def extract_custom_fields(
self, issue_fields: Dict[str, Any]
) -> Dict[str, Any]:
"""Extract custom fields from issue fields dictionary.
Args:
issue_fields: Issue fields dictionary from JIRA
Returns:
Dictionary of custom field names to values
"""
self._load_fields()
custom_fields = {}
if not self._field_cache:
return custom_fields
for field_id, value in issue_fields.items():
if field_id.startswith("customfield_"):
field = self._field_cache.get(field_id)
if field:
custom_fields[field.name] = value
return custom_fields
def clear_cache(self) -> None:
"""Clear the field cache to force reload on next access."""
self._field_cache = None
self._name_to_id_map = None
logger.info("Custom fields cache cleared")