YoutTack to SQLite issues importer 2024

youtrack_issues_importer.py

import argparse
import json
import logging
import os
import sqlite3

import requests
from tqdm import tqdm

# --- Configuration ---
YOUTRACK_URL = "https://issuetracker.YOU-COMPANY.com/api"  # Replace with your YouTrack URL
YOUTRACK_TOKEN = "YOUR TOKEN"  # Replace with your permanent token

DATABASE_FILE = "indeed_youtrack_tasks.db"  # Name of the SQLite database file

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# --- YouTrack API interaction ---
def get_youtrack_issues():
    """Retrieves all issues from YouTrack."""
    if not YOUTRACK_URL or not YOUTRACK_TOKEN:
        raise ValueError("YouTrack URL or token not set in environment variables.")

    headers = {
        "Accept": "application/json",
        "Authorization": f"Bearer {YOUTRACK_TOKEN}"
    }

    BATCH_SIZE = 100
    url = f"{YOUTRACK_URL}/issues"
    params = {
        "$top": BATCH_SIZE,
        "$skip": 0,
        "fields": "idReadable,id,numberInProject,summary,description,created,updated,resolved,updater(login,fullName),creator(login,fullName),project(name,shortName),customFields(name,value(name))"
    }

    all_issues = []
    headers = {
        "Accept": "application/json",
        "Authorization": f"Bearer {YOUTRACK_TOKEN}"
    }
    while True:
        try:
            response = requests.get(url, headers=headers, params=params, timeout=30)
            response.raise_for_status()
            batch = response.json()

            if not batch:  # No more issues to fetch
                break

            all_issues.extend(batch)
            logger.info(f"Fetched {len(batch)} issues, total so far: {len(all_issues)}")

            params["$skip"] += BATCH_SIZE
        except requests.exceptions.RequestException as e:
            logger.error(f"Error fetching issues: {e}")
            break

    return all_issues

def create_database(reset=False):
    """Creates the SQLite database and table."""
    if reset and os.path.exists(DATABASE_FILE):
        os.remove(DATABASE_FILE)

    conn = sqlite3.connect(DATABASE_FILE)
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS issues (
        id TEXT PRIMARY KEY,
        idReadable TEXT,
        project_name TEXT,
        project_shortName TEXT,
        numberInProject INTEGER,
        issue_name TEXT,
        developer_name TEXT,
        planned_for TEXT,
        closed_at INTEGER,
        json_data TEXT
    );
    """)

    conn.commit()
    conn.close()

def save_issues_to_database(issues):
    """Saves the YouTrack issues to the SQLite database."""
    conn = sqlite3.connect(DATABASE_FILE)
    cursor = conn.cursor()

    for issue in tqdm(issues, desc="Saving issues to database"):
        issue_id = issue.get('id')
        idReadable = issue.get('idReadable')
        project = issue.get('project', {})
        project_name = project.get('name')
        project_shortName = project.get('shortName')
        numberInProject = issue.get('numberInProject')
        issue_name = issue.get('summary', '')

        developer_name = ''
        for custom_field in issue.get('customFields', []):
            if custom_field.get('name') == 'Developer':
                value = custom_field.get('value')
                if isinstance(value, dict):
                    developer_name = value.get('name', '')
                break

        if not developer_name:
            for custom_field in issue.get('customFields', []):
                if custom_field.get('name') == 'Assignee':
                    value = custom_field.get('value')
                    if isinstance(value, dict):
                        developer_name = value.get('name', '')
                    break

        planned_for = ''
        for custom_field in issue.get('customFields', []):
            if custom_field.get('name') == 'Planned for':
                value = custom_field.get('value')
                if isinstance(value, dict):
                    planned_for = value.get('name', '')
                break

        closed_at: int|None = issue.get('resolved')
        if closed_at:
            try:
                closed_at: int|None = issue.get('resolved')
            except ValueError:
                logging.warning(f"Invalid date format for issue {issue_id}: {issue.get('resolved')}")

        json_data = json.dumps(issue)

        try:
            cursor.execute("""
                INSERT OR REPLACE INTO issues 
                (id, idReadable, project_name, project_shortName, numberInProject, issue_name, developer_name, planned_for, closed_at, json_data) 
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (issue_id, idReadable, project_name, project_shortName, numberInProject, issue_name, developer_name, planned_for, closed_at, json_data))
        except sqlite3.Error as e:
            logging.error(f"Error inserting issue {issue_id}: {e}")

    conn.commit()
    conn.close()

def main():
    parser = argparse.ArgumentParser(description='YouTrack issues downloader.')
    parser.add_argument('--reset', action='store_true', help='Reset the database before starting.')
    args = parser.parse_args()

    try:
        create_database(reset=args.reset)
        issues = get_youtrack_issues()
        save_issues_to_database(issues)
        print(f"Successfully saved {len(issues)} issues to {DATABASE_FILE}")
    except Exception as e:
        logging.error(f"An error occurred: {e}")

if __name__ == "__main__":
    main()