Source code for langchain_community.utilities.jira

"""Util that calls Jira."""

from typing import Any, Dict, List, Optional

from langchain_core.utils import get_from_dict_or_env
from pydantic import BaseModel, ConfigDict, model_validator


# TODO: think about error handling, more specific api specs, and jql/project limits
[docs] class JiraAPIWrapper(BaseModel): """Wrapper for Jira API.""" jira: Any = None #: :meta private: confluence: Any = None jira_username: Optional[str] = None jira_api_token: Optional[str] = None jira_instance_url: Optional[str] = None jira_cloud: Optional[bool] = None model_config = ConfigDict( extra="forbid", ) @model_validator(mode="before") @classmethod def validate_environment(cls, values: Dict) -> Any: """Validate that api key and python package exists in environment.""" jira_username = get_from_dict_or_env( values, "jira_username", "JIRA_USERNAME", default="" ) values["jira_username"] = jira_username jira_api_token = get_from_dict_or_env( values, "jira_api_token", "JIRA_API_TOKEN" ) values["jira_api_token"] = jira_api_token jira_instance_url = get_from_dict_or_env( values, "jira_instance_url", "JIRA_INSTANCE_URL" ) values["jira_instance_url"] = jira_instance_url jira_cloud_str = get_from_dict_or_env(values, "jira_cloud", "JIRA_CLOUD") jira_cloud = jira_cloud_str.lower() == "true" values["jira_cloud"] = jira_cloud try: from atlassian import Confluence, Jira except ImportError: raise ImportError( "atlassian-python-api is not installed. " "Please install it with `pip install atlassian-python-api`" ) if jira_username == "": jira = Jira( url=jira_instance_url, token=jira_api_token, cloud=jira_cloud, ) else: jira = Jira( url=jira_instance_url, username=jira_username, password=jira_api_token, cloud=jira_cloud, ) confluence = Confluence( url=jira_instance_url, username=jira_username, password=jira_api_token, cloud=jira_cloud, ) values["jira"] = jira values["confluence"] = confluence return values
[docs] def parse_issues(self, issues: Dict) -> List[dict]: parsed = [] for issue in issues["issues"]: key = issue["key"] summary = issue["fields"]["summary"] created = issue["fields"]["created"][0:10] priority = issue["fields"]["priority"]["name"] status = issue["fields"]["status"]["name"] try: assignee = issue["fields"]["assignee"]["displayName"] except Exception: assignee = "None" rel_issues = {} for related_issue in issue["fields"]["issuelinks"]: if "inwardIssue" in related_issue.keys(): rel_type = related_issue["type"]["inward"] rel_key = related_issue["inwardIssue"]["key"] rel_summary = related_issue["inwardIssue"]["fields"]["summary"] if "outwardIssue" in related_issue.keys(): rel_type = related_issue["type"]["outward"] rel_key = related_issue["outwardIssue"]["key"] rel_summary = related_issue["outwardIssue"]["fields"]["summary"] rel_issues = {"type": rel_type, "key": rel_key, "summary": rel_summary} parsed.append( { "key": key, "summary": summary, "created": created, "assignee": assignee, "priority": priority, "status": status, "related_issues": rel_issues, } ) return parsed
[docs] def parse_projects(self, projects: List[dict]) -> List[dict]: parsed = [] for project in projects: id = project["id"] key = project["key"] name = project["name"] type = project["projectTypeKey"] style = project["style"] parsed.append( {"id": id, "key": key, "name": name, "type": type, "style": style} ) return parsed
[docs] def search(self, query: str) -> str: issues = self.jira.jql(query) parsed_issues = self.parse_issues(issues) parsed_issues_str = ( "Found " + str(len(parsed_issues)) + " issues:\n" + str(parsed_issues) ) return parsed_issues_str
[docs] def project(self) -> str: projects = self.jira.projects() parsed_projects = self.parse_projects(projects) parsed_projects_str = ( "Found " + str(len(parsed_projects)) + " projects:\n" + str(parsed_projects) ) return parsed_projects_str
[docs] def issue_create(self, query: str) -> str: try: import json except ImportError: raise ImportError( "json is not installed. Please install it with `pip install json`" ) params = json.loads(query) return self.jira.issue_create(fields=dict(params))
[docs] def page_create(self, query: str) -> str: try: import json except ImportError: raise ImportError( "json is not installed. Please install it with `pip install json`" ) params = json.loads(query) return self.confluence.create_page(**dict(params))
[docs] def other(self, query: str) -> str: try: import json except ImportError: raise ImportError( "json is not installed. Please install it with `pip install json`" ) params = json.loads(query) jira_function = getattr(self.jira, params["function"]) return jira_function(*params.get("args", []), **params.get("kwargs", {}))
[docs] def run(self, mode: str, query: str) -> str: if mode == "jql": return self.search(query) elif mode == "get_projects": return self.project() elif mode == "create_issue": return self.issue_create(query) elif mode == "other": return self.other(query) elif mode == "create_page": return self.page_create(query) else: raise ValueError(f"Got unexpected mode {mode}")