### ===================================================================================================================
### JIRA REST API functions
### ===================================================================================================================
# Copyright ©VIIA 2025
### ===================================================================================================================
### 1. Import modules
### ===================================================================================================================
# General imports
from __future__ import annotations
from typing import TYPE_CHECKING, Dict, Optional
import requests
from requests import Response
import sys
import os
import json
import yaml
# References for functions and classes in the viiaPackage
if TYPE_CHECKING:
from viiapackage.viiaStatus import ViiaProject
# Collect the connection settings to connect to MYVIIA
if os.environ.get('USERNAME_JIRA') is None or os.environ.get('TOKEN_JIRA') is None:
try:
from user_config import connection_dict
if 'jira' in connection_dict.keys():
os.environ['USERNAME_JIRA'] = connection_dict['jira']['email']
os.environ['TOKEN_JIRA'] = connection_dict['jira']['token']
use_user_config = True
else:
use_user_config = False
except ImportError:
connection_dict = dict()
use_user_config = False
else:
use_user_config = True
### ===================================================================================================================
### 2. Helper functions
### ===================================================================================================================
def _get_viia_load_jira_stories(project: ViiaProject) -> Optional[Dict]:
""" Helper function to get the values for the JIRA stories."""
# Read yaml with JIRA stories of VIIA project
with open(project.viia_settings.project_specific_package_location / 'tools' / 'jira_stories.yaml') as f:
return yaml.load(f, Loader=yaml.FullLoader)
### ===================================================================================================================
### 3. Function to create the VIIA object in JIRA
### ===================================================================================================================
[docs]def viia_create_object_jira(
project: ViiaProject, jira_board: str, object_nr: Optional[str] = None, analysis_type: Optional[str] = None,
object_size: Optional[str] = None, assignee_email: Optional[str] = None):
"""
This function connects with JIRA REST API in order to create object tasks on VIIA JIRA boards. JIRA credentials in
the user_config are required for this function The following entry is expected in the user_config connection_dict:
.. code-block:: python
{'jira': {
'email': 'employee.lastname@rhdhv.com',
'token': 'abc123XYZ'}}
The API token can be created on the JIRA website in the 'Security' tab of account settings:
https://id.atlassian.com/manage-profile/security/api-tokens
Input:
- project (obj): ViiaProject to retrieve project data from
- jira_board (str): Referring to the board of the corresponding production team. Choose from 'DIANA1', 'DIANA2',
'DIANA3', 'DIANA4', 'DIANA5' and 'DIANA6'.
- object_nr (str): VIIA object number.
- analysis_type (str): Select the analysis type for the object, for example 'NLTH'.
- object_size (str): Select from 'Small', 'Medium' and 'Large'.
- Optional: assignee_email (str): Email-adress with which the desired assignee is registered at the JIRA board.
Output:
- An epic and all user stories for the object will be created in the requested JIRA board, including components
and story points, and assignee if selected.
"""
def handle_response(response: Response) -> dict:
""" This function checks the validity of the response. Standardised status codes are interpreted and logged to
the project."""
if response.status_code == 200:
project.write_log(f"Request accepted")
return response.json()
elif response.status_code == 201:
project.write_log(f"Successfully created {response.json()['key']} at JIRA")
return response.json()
elif response.status_code == 400:
project.write_log(
f"Sent request to JIRA REST API server, but found an issue with input. Reason: {response.text}")
return response.json()
elif response.status_code == 404:
project.write_log(f"ERROR: Requested API not found, check URL: {response.request.url}")
return {}
elif response.status_code == 401:
project.write_log(f"ERROR: Authentication to JIRA REST API server failed. Please check your credentials")
return {}
elif not response:
project.write_log(f"ERROR: No connection to api server @{response.request.url}")
return {}
else:
project.write_log(
f"Request to api server @{url} failed with status code {response.status_code}. Reason: "
f"{response.reason}")
return response.json()
# Checking whether function is called from testing module
if sys._getframe(1).f_code.co_name == 'test_viia_jira':
test = True
else:
test = False
# Input handling
if object_nr and object_nr != project.name and (
not object_size or not (analysis_type or project.project_information['analysis_type'])):
project.write_log(
"ERROR: When using a custom object_nr, inputting analysis_type and object_size is compulsary.")
return
if not object_nr:
object_nr = project.name
if not analysis_type:
if project.project_information['analysis_type']:
analysis_type = project.project_information['analysis_type']
elif [objectdeel for objectdeel in project.project_information['objectdelen'] if
objectdeel['naam'] == project.project_information['objectdeel']][0]['rekenmethodiek_id']:
analysis_type = [objectdeel for objectdeel in project.project_information['objectdelen'] if
objectdeel['naam'] == project.project_information['objectdeel']][0]['rekenmethodiek_id'][
'rekenmethodiek']
if analysis_type and analysis_type.upper() not in [
'NLTH', 'SBS-1', 'SBS-2', 'NLPO', 'MRS', 'NLTH-UPDATE', 'NLTH-TVA-UPDATE', 'NLTH-PARA', 'REF', 'REF-NLPO',
'REF-AGRO']:
if analysis_type.upper() in ['NLPO', 'SLAMA', 'REF']:
project.write_log(
f"ERROR: Analysis type {analysis_type} is currently not available for JIRA. If required, submit a "
f"feature request at VIIA Automation.")
return
else:
project.write_log(f"ERROR: Analysis type {analysis_type} is unknown. Please check/use manual input.")
return
elif not analysis_type:
project.write_log(f"ERROR: No valid analysis type could be found. Please check/use manual input.")
return
if not object_size:
object_size_id = [objectdeel for objectdeel in project.project_information['objectdelen'] if
objectdeel['naam'] == project.project_information['objectdeel']][0]['object_deel_grootte_id']
object_size = {'1': 'Small', '2': 'Medium', '3': 'Large'}[str(object_size_id)]
elif object_size.lower() in ['small', 'medium', 'large']:
object_size = {'small': 'Small', 'medium': 'Medium', 'large': 'Large'}[object_size.lower()]
else:
project.write_log(f"ERROR: No valid object size could be found. Please check/use manual input.")
return
if jira_board not in ['DIANA1', 'DIANA2', 'DIANA3', 'DIANA4', 'DIANA5', 'DIANA6']:
project.write_log(
f"ERROR: JIRA board {jira_board} is not known. Choose from 'DIANA1', 'DIANA2', 'DIANA3', 'DIANA4', "
f"'DIANA5' or 'DIANA6'. Please check/use manual input.")
return
# Acquiring preset story data for JIRA
template_data = _get_viia_load_jira_stories(project=project)
headers = {
'Accept': 'application/json',
'Content-Type': 'application/json'}
# Retrieving userID of the assignee, if requested
if assignee_email:
url_assignee = f"https://royalhaskoningdhv.atlassian.net/rest/api/2/user/search?query={assignee_email}"
assignee_response = requests.get(
url_assignee, headers=headers, auth=(os.environ['USERNAME_JIRA'], os.environ['TOKEN_JIRA']))
assignee_data = handle_response(assignee_response)
if assignee_data:
url_users = \
f"https://royalhaskoningdhv.atlassian.net/rest/api/2/project/" \
f"{template_data['Boards'][jira_board]['key']}/role/10100"
users_response = requests.get(
url_users, headers=headers, auth=(os.environ['USERNAME_JIRA'], os.environ['TOKEN_JIRA']))
users_data = handle_response(users_response)
if not users_data:
project.write_log(
f"Could not retrieve user ID's from board {template_data['Boards'][jira_board]['key']}."
f" Continuing without user assignment")
assignee_id = None
elif assignee_data[0]['accountId'] in [actor['actorUser']['accountId'] for actor in users_data['actors']]:
assignee_id = assignee_data[0]['accountId']
else:
project.write_log(
f"WARNING: {assignee_data[0]['displayName']} is not a user on board "
f"{template_data['Boards'][jira_board]['key']}. Continuing without user assignment")
assignee_id = None
else:
project.write_log(
f"WARNING: could not find user ID for selected assignee email '{assignee_email}'."
f" Continuing without user assigment")
assignee_id = None
else:
assignee_id = None
# Preparing and creating the Epic
url = "https://royalhaskoningdhv.atlassian.net/rest/api/2/issue"
payload_epic = {'fields': {
'project': {
'key': template_data['Boards'][jira_board]['key']},
'issuetype': {
'name': 'Epic'},
'summary': f'{object_nr} | {analysis_type} | {object_size}',
'customfield_10005': f'{object_nr} | {analysis_type} | {object_size}'}}
if assignee_id:
payload_epic['fields']['assignee'] = {"accountId": f"{assignee_id}"}
payload_epic = json.dumps(payload_epic)
epic_response = None
if not test:
epic_response = requests.post(
url, headers=headers, data=payload_epic, auth=(os.environ['USERNAME_JIRA'], os.environ['TOKEN_JIRA']))
epic_data = handle_response(epic_response)
if epic_response.status_code not in [200, 201]:
project.write_log(f"ERROR: could not create requested epic. Aborting function")
return
else:
# An artificial epic key is assigned for testing purposes
epic_data = {'key': 'TEST'}
# Preparing and creating the user stories
payload_story = None
for story in template_data['Stories'][analysis_type].values():
payload_story = {
'fields': {
'project': {
'key': template_data['Boards'][jira_board]['key']},
'summary': story['Summary'],
'customfield_10008': epic_data['key'],
'issuetype': {
'name': 'Story'},
'customfield_10032':
story['Story_points'] * template_data['Story_point_multipliers'][analysis_type][object_size],
'components': [{'name': component} for component in story['Components']]}}
if assignee_id:
payload_story['fields']['assignee'] = {'accountId': f'{assignee_id}'}
payload_story = json.dumps(payload_story)
if not test:
story_response = requests.post(
url, headers=headers, data=payload_story, auth=(os.environ['USERNAME_JIRA'], os.environ['TOKEN_JIRA']))
handle_response(story_response)
if epic_response.status_code not in [200, 201]:
project.write_log(f"ERROR: could not create requested user story. Aborting function")
return
if test:
return json.loads(payload_epic), json.loads(payload_story)
else:
project.write_log(f"Object creation in JIRA completed")
return
### ===================================================================================================================
### 4. End of script
### ===================================================================================================================