Introduction: In a previous post (Automate Cleaning Up Duplicate Project Roles After Migrating to Atlassian Cloud), I explained how to automate the cleanup of duplicate project roles after migration. However, before running that script, it’s essential to verify if your workflows depend on any project roles that will no longer exist after the merge. This step is vital to prevent any potential issues that could break your workflows once the roles are merged or deleted.

Why This Script is Important: Merging or cleaning up project roles without checking their usage in workflows can lead to broken transitions, conditions, validators, or post functions, causing your workflows to fail. This script will help you identify any workflows that are currently using specific project roles in conditions, validators, or post functions, so you can update or modify them before proceeding with the merge.

How the Script Works: This Python script scans all workflows in your Jira instance, focusing on transitions and their associated rules—conditions, validators, and post functions. It looks for any dependencies on project roles and exports this information into two CSV files:

  1. workflow_transition_rules.csv – Contains all the details of conditions, validators, and post functions within transitions.
  2. special_conditions_functions.csv – Highlights specific conditions and functions that involve project roles, which require particular attention.

Configure the Script:

  • Open the script in your preferred text editor.
  • Update the config dictionary with your Jira instance details:
    • email: Your Jira account email.
    • token: Your Jira API token.
    • base_url: The base URL of your Jira instance, e.g., https://yourdomain.atlassian.net.

Script Breakdown:

import os
import requests
import logging
import csv
from requests.auth import HTTPBasicAuth
from concurrent.futures import ThreadPoolExecutor, as_completed

# Logging configuration
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s: %(message)s')

# Authentication and base API URL settings
config = {
    'email': '',  
    'token': '',  
    'base_url': 'https://domain.atlassian.net',
}

# Output CSV file paths
OUTPUT_CSV = 'workflow_transition_rules.csv'
SPECIAL_CSV = 'special_conditions_functions.csv'
PAGE_SIZE = 50  # Number of items per page
MAX_WORKERS = 10  # Maximum number of concurrent threads

# Function to fetch a page of workflows
def fetch_workflows_page(start_at):
    url = f"{config['base_url']}/rest/api/2/workflow/search"
    auth = HTTPBasicAuth(config['email'], config['token'])
    params = {
        "expand": "transitions.rules",
        "startAt": start_at,
        "maxResults": PAGE_SIZE
    }
    response = requests.get(url, auth=auth, params=params, headers={"Accept": "application/json"})

    if response.status_code != 200:
        logging.error(f"Error fetching workflows at page {start_at}: {response.status_code} - {response.text}")
        return []

    data = response.json()
    return data.get('values', []), data.get('isLast', True)

# Function to fetch all workflows using multithreading
def get_all_workflows():
    all_workflows = []
    start_at = 0
    is_last = False

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        while not is_last:
            futures = [executor.submit(fetch_workflows_page, start_at)]
            for future in as_completed(futures):
                workflows, is_last = future.result()
                all_workflows.extend(workflows)
                start_at += PAGE_SIZE

    return all_workflows

# Function to process workflows and save transition rules to CSVs
def process_workflows_and_save_details():
    workflows = get_all_workflows()
    
    logging.info(f"Total workflows found: {len(workflows)}")

    with open(OUTPUT_CSV, mode='w', newline='') as output_file, \
         open(SPECIAL_CSV, mode='w', newline='') as special_file:
        
        writer = csv.writer(output_file)
        special_writer = csv.writer(special_file)
        
        writer.writerow([
            'Workflow Name', 'Workflow ID', 'Transition ID', 'Transition Name', 
            'Condition Type', 'Condition Configuration', 'Validator Type', 
            'Validator Configuration', 'Post Function Type', 
            'Post Function Configuration'
        ])
        
        special_writer.writerow([
            'Workflow Name', 'Workflow ID', 'Transition ID', 'Transition Name', 
            'Condition Type', 'Condition Configuration', 'Validator Type', 
            'Validator Configuration', 'Post Function Type', 
            'Post Function Configuration'
        ])

        excluded_functions = {
            'IssueReindexFunction', 
            'GenerateChangeHistoryFunction', 
            'CreateCommentFunction', 
            'UpdateIssueStatusFunction', 
            'FireIssueEventFunction',
            'IssueCreateFunction'
        }

        special_conditions = {'InProjectRoleCondition'}
        special_functions = {'SetIssueSecurityFromRoleFunction'}

        for workflow in workflows:
            workflow_name = workflow.get('id', {}).get('name', 'Unknown')
            workflow_id = workflow.get('id', {}).get('entityId', 'Unknown')

            # Process transitions
            for transition in workflow.get('transitions', []):
                transition_id = transition.get('id', 'Unknown')
                transition_name = transition.get('name', 'Unknown')

                # Extract rules: conditions, validators, and post functions
                conditions = transition.get('rules', {}).get('conditionsTree', {})
                validators = transition.get('rules', {}).get('validators', [])
                post_functions = transition.get('rules', {}).get('postFunctions', [])

                should_write_to_special = False

                # Process conditions
                if conditions:
                    if conditions.get('nodeType') == 'simple':
                        condition_type = conditions.get('type', 'Unknown')
                        condition_config = conditions.get('configuration', {})
                    else:
                        condition_type = "Complex"
                        condition_config = conditions

                    if condition_type in special_conditions:
                        should_write_to_special = True

                    for validator in validators:
                        validator_type = validator.get('type', 'Unknown')
                        validator_config = validator.get('configuration', {})

                        for post_function in post_functions:
                            post_function_type = post_function.get('type', 'Unknown')
                            if post_function_type in excluded_functions:
                                continue
                            if post_function_type in special_functions:
                                should_write_to_special = True
                            post_function_config = post_function.get('configuration', {})

                            writer.writerow([
                                workflow_name, workflow_id, transition_id, transition_name, 
                                condition_type, str(condition_config), validator_type, 
                                str(validator_config), post_function_type, 
                                str(post_function_config)
                            ])

                            if should_write_to_special:
                                special_writer.writerow([
                                    workflow_name, workflow_id, transition_id, transition_name, 
                                    condition_type, str(condition_config), validator_type, 
                                    str(validator_config), post_function_type, 
                                    str(post_function_config)
                                ])
                    # If no validators, still need to capture post functions
                    if not validators:
                        for post_function in post_functions:
                            post_function_type = post_function.get('type', 'Unknown')
                            if post_function_type in excluded_functions:
                                continue
                            if post_function_type in special_functions:
                                should_write_to_special = True
                            post_function_config = post_function.get('configuration', {})

                            writer.writerow([
                                workflow_name, workflow_id, transition_id, transition_name, 
                                condition_type, str(condition_config), '', '', 
                                post_function_type, str(post_function_config)
                            ])

                            if should_write_to_special:
                                special_writer.writerow([
                                    workflow_name, workflow_id, transition_id, transition_name, 
                                    condition_type, str(condition_config), '', '', 
                                    post_function_type, str(post_function_config)
                                ])
                else:
                    for validator in validators:
                        validator_type = validator.get('type', 'Unknown')
                        validator_config = validator.get('configuration', {})

                        for post_function in post_functions:
                            post_function_type = post_function.get('type', 'Unknown')
                            if post_function_type in excluded_functions:
                                continue
                            if post_function_type in special_functions:
                                should_write_to_special = True
                            post_function_config = post_function.get('configuration', {})

                            writer.writerow([
                                workflow_name, workflow_id, transition_id, transition_name, 
                                '', '', validator_type, str(validator_config), 
                                post_function_type, str(post_function_config)
                            ])

                            if should_write_to_special:
                                special_writer.writerow([
                                    workflow_name, workflow_id, transition_id, transition_name, 
                                    '', '', validator_type, str(validator_config), 
                                    post_function_type, str(post_function_config)
                                ])
                    # If no validators, still need to capture post functions
                    if not validators:
                        for post_function in post_functions:
                            post_function_type = post_function.get('type', 'Unknown')
                            if post_function_type in excluded_functions:
                                continue
                            if post_function_type in special_functions:
                                should_write_to_special = True
                            post_function_config = post_function.get('configuration', {})

                            writer.writerow([
                                workflow_name, workflow_id, transition_id, transition_name, 
                                '', '', '', '', post_function_type, str(post_function_config)
                            ])

                            if should_write_to_special:
                                special_writer.writerow([
                                    workflow_name, workflow_id, transition_id, transition_name, 
                                    '', '', '', '', post_function_type, str(post_function_config)
                                ])

# Run the script
if __name__ == "__main__":
    try:
        process_workflows_and_save_details()
        logging.info(f"Transition rules have been saved in {OUTPUT_CSV} and {SPECIAL_CSV}")
    except Exception as e:
        logging.error(f"An error occurred: {str(e)}")

Run the Script: To execute the script, run the following command in your terminal:

python your_script_name.py

Replace your_script_name.py with the actual name of your Python script file.

Conclusion: Before proceeding with the merging or cleanup of project roles, it’s crucial to run this script to identify any workflows that depend on these roles. This will ensure that your workflows remain intact, avoiding potential disruptions in your processes. Once you have updated the necessary workflows, you can confidently execute the role cleanup script discussed in the previous post.

Feel free to share your thoughts or any challenges you face in the comments section below!