ops_utils.jira_util

  1"""Module for interacting with Jira tickets."""
  2import os
  3import logging
  4from jira 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 Jira:
 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 = 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(server=self.server, basic_auth=(jira_user, 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        issue = self.jira.issue(issue_key)
 75        issue.update(fields=field_update_dict)
 76
 77    def add_comment(self, issue_key: str, comment: str) -> None:
 78        """
 79        Add a comment to a Jira ticket.
 80
 81        **Args:**
 82        - issue_key (str): The issue key to update
 83        - comment (str): The comment to add
 84        """
 85        self.jira.add_comment(issue_key, comment)
 86
 87    def transition_ticket(self, issue_key: str, transition_id: int) -> None:
 88        """
 89        Transition a Jira ticket to a new status.
 90
 91        **Args:**
 92        - issue_key (str): The issue key to update
 93        - transition_id (int): The status ID to transition the issue to
 94        """
 95        self.jira.transition_issue(issue_key, transition_id)
 96
 97    def get_issues_by_criteria(
 98            self,
 99            criteria: str,
100            max_results: int = 200,
101            fields: Optional[list[str]] = None,
102            expand_info: Optional[str] = None
103    ) -> list[dict]:
104        """
105        Get all issues by defining specific criteria.
106
107        **Args:**
108        - criteria (str): The criteria to search for. This should be formatted in a supported JIRA search filter
109        (i.e. `project = '{project}' AND sprint = '{sprint_name}' AND status = {status}`)
110        - max_results (int): The maximum number of results to return. Defaults to 200.
111        - fields (list[string], optional): The fields to include in the return for each
112        ticket (i.e. `["summary", "status", "assignee"]`)
113
114        **Returns:**
115        - list[dict]: The list of issues matching the criteria
116        """
117        logging.info(f"Getting issues by criteria: {criteria}")
118        if fields:
119            return self.jira.search_issues(criteria, fields=fields, maxResults=max_results, expand=expand_info)
120        return self.jira.search_issues(criteria, maxResults=max_results, expand=expand_info)