"""Util that calls Jira."""
from typing import Any, Dict, List, Optional, Union
from langchain_core.utils import get_from_dict_or_env
from pydantic import BaseModel, ConfigDict, model_validator
from typing_extensions import TypedDict
[docs]
class JiraOauth2Token(TypedDict):
"""Jira OAuth2 token."""
access_token: str
"""Jira OAuth2 access token."""
token_type: str
"""Jira OAuth2 token type ('bearer' or other)."""
[docs]
class JiraOauth2(TypedDict):
"""Jira OAuth2."""
client_id: str
"""Jira OAuth2 client ID."""
token: JiraOauth2Token
"""Jira OAuth2 token."""
# TODO: think about error handling, more specific api specs, and jql/project limits
[docs]
class JiraAPIWrapper(BaseModel):
"""
Wrapper for Jira API. You can connect to Jira with either an API token or OAuth2.
- with API token, you need to provide the JIRA_USERNAME and JIRA_API_TOKEN
environment variables or arguments.
ex: JIRA_USERNAME=your_username JIRA_API_TOKEN=your_api_token
- with OAuth2, you need to provide the JIRA_OAUTH2 environment variable or
argument as a dict having as fields "client_id" and "token" which is
a dict containing at least "access_token" and "token_type".
ex: JIRA_OAUTH2='{"client_id": "your_client_id", "token":
{"access_token": "your_access_token","token_type": "bearer"}}'
"""
jira: Any = None #: :meta private:
confluence: Any = None
jira_username: Optional[str] = None
jira_api_token: Optional[str] = None
"""Jira API token when you choose to connect to Jira with api token."""
jira_oauth2: Optional[Union[JiraOauth2, str]] = None
"""Jira OAuth2 token when you choose to connect to Jira with oauth2."""
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", default=""
)
values["jira_api_token"] = jira_api_token
jira_oauth2 = get_from_dict_or_env(
values, "jira_oauth2", "JIRA_OAUTH2", default=""
)
values["jira_oauth2"] = jira_oauth2
if jira_oauth2 and isinstance(jira_oauth2, str):
try:
import json
jira_oauth2 = json.loads(jira_oauth2)
except ImportError:
raise ImportError(
"json is not installed. Please install it with `pip install json`"
)
except json.decoder.JSONDecodeError as e:
raise ValueError(
f"The format of the JIRA_OAUTH2 string is "
f"not a valid dictionary: {e}"
)
jira_instance_url = get_from_dict_or_env(
values, "jira_instance_url", "JIRA_INSTANCE_URL"
)
values["jira_instance_url"] = jira_instance_url
if "jira_cloud" in values and values["jira_cloud"] is not None:
values["jira_cloud"] = str(values["jira_cloud"])
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
if jira_api_token and jira_oauth2:
raise ValueError(
"You have to provide either a jira_api_token or a jira_oauth2. "
"Not both."
)
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_api_token:
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,
)
elif jira_oauth2:
jira = Jira(
url=jira_instance_url,
oauth2=jira_oauth2,
cloud=jira_cloud,
)
confluence = Confluence(
url=jira_instance_url,
oauth2=jira_oauth2,
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]
if "priority" in issue["fields"]:
priority = issue["fields"]["priority"]["name"]
else:
priority = None
status = issue["fields"]["status"]["name"]
try:
assignee = issue["fields"]["assignee"]["displayName"]
except Exception:
assignee = "None"
rel_issues = {}
for related_issue in issue["fields"].get("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.get("projectTypeKey")
style = project.get("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}")
