ops_utils.jira_util

  1"""Module for interacting with Jira tickets."""
  2import os
  3import logging
  4from atlassian import Jira
  5from google.cloud import secretmanager
  6from typing import Optional
  7
  8WORKBENCH_SERVER = "https://broadworkbench.atlassian.net/"
  9"""The default Broad workbench server address"""
 10BROAD_INSTITUTE_SERVER = "https://broadinstitute.atlassian.net/"
 11"""The default Broad Jira server address"""
 12
 13
 14class JiraUtil:
 15    """
 16    A class to assist in interacting with JIRA tickets.
 17
 18    Provides functionality to help in updating tickets
 19    (adding comments, updating ticket fields, and transitioning statuses). Also provides a way to query
 20    existing JIRA tickets using certain filters. Assumes that an accessible JIRA API key is stored in
 21    Google's SecretManger
 22    """
 23
 24    def __init__(self, server: str, gcp_project_id: str, jira_api_key_secret_name: str) -> None:
 25        """
 26        Initialize the Jira instance using the provided server.
 27
 28        **Args:**
 29        - server (str): The server URL to connect to. For example: `https://broadinstitute.atlassian.net/`
 30        - gcp_project_id (str): The GCP project ID used to locate the Jira API key that is stored in SecretManager.
 31        - jira_api_key_secret_name (str): The name of the Jira API key secret that is stored in SecretManager.
 32        """
 33        self.server = server
 34        """@private"""
 35        self.gcp_project_id = gcp_project_id
 36        """@private"""
 37        self.jira_api_key_secret_name = jira_api_key_secret_name
 38        """@private"""
 39        self.jira_connection = self._connect_to_jira()
 40        """@private"""
 41
 42    def _connect_to_jira(self) -> Jira:
 43        """
 44        Obtain credentials and establish the Jira connection.
 45
 46        User must have token stored in ~/.jira_api_key
 47        """
 48        if os.getenv("RUN_IN_CLOUD") == "yes":
 49            jira_user = f'{os.getenv("JIRA_USER")}@broadinstitute.org'
 50        else:
 51            jira_user = f'{os.getenv("USER")}@broadinstitute.org'
 52
 53        # This is necessary for when scripts are using this utility in a Cloud Function
 54        try:
 55            jira_api_token_file_path = os.path.expanduser("~/.jira_api_key")
 56            with open(jira_api_token_file_path, "r") as token_file:
 57                token = token_file.read().strip()
 58        except FileNotFoundError:
 59            client = secretmanager.SecretManagerServiceClient()
 60            name = f"projects/{self.gcp_project_id}/secrets/{self.jira_api_key_secret_name}/versions/latest"
 61            token = client.access_secret_version(name=name).payload.data.decode("UTF-8")
 62
 63        return Jira(url=self.server, username=jira_user, password=token)
 64
 65    def update_ticket_fields(self, issue_key: str, field_update_dict: dict) -> None:
 66        """
 67        Update a Jira ticket with new field values.
 68
 69        **Args:**
 70        - issue_key (str): The issue key to update
 71        - field_update_dict (dict): The field values to update. Formatted with the field ID as the key,
 72        and the updated value as the key's value.
 73        """
 74        self.jira_connection.issue_update(issue_key, field_update_dict)
 75
 76    def add_comment(self, issue_key: str, comment: str) -> None:
 77        """
 78        Add a comment to a Jira ticket.
 79
 80        **Args:**
 81        - issue_key (str): The issue key to update
 82        - comment (str): The comment to add
 83        """
 84        self.jira_connection.issue_add_comment(issue_key, comment)
 85
 86    def transition_ticket(self, issue_key: str, transition_id: int) -> None:
 87        """
 88        Transition a Jira ticket to a new status.
 89
 90        **Args:**
 91        - issue_key (str): The issue key to update
 92        - transition_id (int): The status ID to transition the issue to
 93        """
 94        self.jira_connection.set_issue_status_by_transition_id(issue_key, transition_id)
 95
 96    def get_issues_by_criteria(
 97            self,
 98            criteria: str,
 99            max_results: int = 200,
100            fields: Optional[list[str]] = None,
101            expand_info: Optional[str] = None
102    ) -> list[dict]:
103        """
104        Get all issues by defining specific criteria.
105
106        **Args:**
107        - criteria (str): The criteria to search for. This should be formatted in a supported JIRA search filter
108        (i.e. `project = '{project}' AND sprint = '{sprint_name}' AND status = {status}`)
109        - max_results (int): The maximum number of results to return. Defaults to 200.
110        - fields (list[string], optional): A list of the fields to include in the return for each
111        ticket (i.e. `["summary", "status", "assignee"]`)
112
113        **Returns:**
114        - list[dict]: The list of issues matching the criteria
115        """
116        logging.info(f"Getting issues by criteria: {criteria}")
117
118        payload = {
119            "jql": criteria,
120            "maxResults": max_results
121        }
122
123        if fields:
124            payload["fields"] = fields
125        if expand_info:
126            payload["expand"] = expand_info
127
128        return self.jira_connection.post("rest/api/3/search/jql", data=payload)