export_linear_issues.py

#!/usr/bin/env python3
"""
Linear Issues Export Script

This script retrieves all issues closed in the last 12 months from Linear
and exports them to a CSV file for audit evidence.

Copyright (c) 2025 24Slides
Licensed under the MIT License - see LICENSE file for details.
"""

import os
import sys
import csv
import requests
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional

# Try to load .env file if python-dotenv is available
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass  # python-dotenv not installed, will use system environment variables

# Configuration constants
LINEAR_API_URL = "https://api.linear.app/graphql"
ISSUES_PER_PAGE = 100
OUTPUT_FILENAME = "linear_closed_issues_last_12mo.csv"
MONTHS_BACK = 12
MAX_RETRIES = 5
INITIAL_RETRY_DELAY = 1  # seconds


def get_twelve_months_ago() -> str:
    """
    Calculate the date exactly 12 months ago from today.
    
    Returns ISO 8601 formatted datetime string that Linear's API accepts.
    """
    today = datetime.now()
    twelve_months_ago = today - timedelta(days=365)
    # Return in ISO 8601 format with 'Z' suffix for UTC
    return twelve_months_ago.strftime("%Y-%m-%dT%H:%M:%S.000Z")


def safe_get_nested(data: Dict, *keys: str, default: str = "") -> str:
    """
    Safely get a nested dictionary value.
    
    Args:
        data: The dictionary to search
        *keys: The keys to traverse
        default: Default value if key doesn't exist
        
    Returns:
        The value if found, otherwise the default
    """
    current = data
    for key in keys:
        if current is None or not isinstance(current, dict):
            return default
        current = current.get(key)
    return current if current is not None else default


def make_graphql_request(url: str, headers: Dict, query: str, variables: Dict, max_retries: int = MAX_RETRIES) -> Dict:
    """
    Make a GraphQL request with exponential backoff retry logic for rate limiting.
    
    Args:
        url: GraphQL endpoint URL
        headers: Request headers including authorization
        query: GraphQL query string
        variables: Query variables
        max_retries: Maximum number of retry attempts
        
    Returns:
        JSON response data
        
    Raises:
        requests.exceptions.RequestException: If request fails after all retries
        ValueError: If response contains GraphQL errors
    """
    for attempt in range(max_retries):
        try:
            response = requests.post(
                url,
                json={"query": query, "variables": variables},
                headers=headers,
                timeout=30
            )
            
            # Handle rate limiting (HTTP 429)
            if response.status_code == 429:
                if attempt < max_retries - 1:
                    retry_delay = INITIAL_RETRY_DELAY * (2 ** attempt)  # Exponential backoff
                    print(f"Rate limit hit. Retrying in {retry_delay} seconds... (Attempt {attempt + 1}/{max_retries})")
                    time.sleep(retry_delay)
                    continue
                else:
                    raise requests.exceptions.RequestException("Rate limit exceeded after maximum retries")
            
            # Handle other non-200 status codes
            if response.status_code != 200:
                error_msg = f"API request failed with status {response.status_code}: {response.text}"
                raise requests.exceptions.RequestException(error_msg)
            
            data = response.json()
            
            # Check for GraphQL errors
            if "errors" in data:
                error_msg = f"GraphQL errors: {data['errors']}"
                raise ValueError(error_msg)
            
            # Validate response structure
            if "data" not in data:
                raise ValueError("Invalid response structure from Linear API")
            
            return data
            
        except requests.exceptions.Timeout:
            if attempt < max_retries - 1:
                retry_delay = INITIAL_RETRY_DELAY * (2 ** attempt)
                print(f"Request timeout. Retrying in {retry_delay} seconds... (Attempt {attempt + 1}/{max_retries})")
                time.sleep(retry_delay)
                continue
            else:
                raise
        except requests.exceptions.ConnectionError:
            if attempt < max_retries - 1:
                retry_delay = INITIAL_RETRY_DELAY * (2 ** attempt)
                print(f"Connection error. Retrying in {retry_delay} seconds... (Attempt {attempt + 1}/{max_retries})")
                time.sleep(retry_delay)
                continue
            else:
                raise
    
    raise requests.exceptions.RequestException("Failed after maximum retries")


def fetch_closed_issues(api_key: str) -> List[Dict]:
    """
    Fetch all closed issues from Linear API with pagination.
    
    Args:
        api_key: Linear API key for authentication
        
    Returns:
        List of issues with all required fields
        
    Raises:
        requests.exceptions.RequestException: If API request fails
        ValueError: If API returns invalid data
    """
    headers = {
        "Authorization": api_key,
        "Content-Type": "application/json"
    }
    
    twelve_months_ago = get_twelve_months_ago()
    
    # GraphQL query with pagination support
    query = """
    query ClosedIssues($cursor: String, $completedAfter: DateTimeOrDuration!, $first: Int!) {
      issues(
        first: $first
        after: $cursor
        filter: {
          completedAt: { gte: $completedAfter }
          state: { type: { in: ["completed", "canceled"] } }
        }
      ) {
        nodes {
          identifier
          title
          completedAt
          url
          creator {
            name
          }
          assignee {
            name
          }
          team {
            name
          }
        }
        pageInfo {
          hasNextPage
          endCursor
        }
      }
    }
    """
    
    all_issues = []
    has_next_page = True
    cursor = None
    page_count = 0
    
    print(f"Fetching issues completed after: {twelve_months_ago}")
    print("=" * 60)
    
    while has_next_page:
        variables = {
            "cursor": cursor,
            "completedAfter": twelve_months_ago,
            "first": ISSUES_PER_PAGE
        }
        
        # Make request with retry logic
        data = make_graphql_request(LINEAR_API_URL, headers, query, variables)
        
        if "issues" not in data["data"]:
            raise ValueError("Invalid response structure from Linear API")
        
        issues_data = data["data"]["issues"]
        issues = issues_data["nodes"]
        page_info = issues_data["pageInfo"]
        
        all_issues.extend(issues)
        page_count += 1
        
        print(f"Page {page_count}: Fetched {len(issues)} issues (Total: {len(all_issues)})")
        
        has_next_page = page_info["hasNextPage"]
        cursor = page_info["endCursor"]
    
    print("=" * 60)
    print(f"Total issues fetched: {len(all_issues)}")
    
    return all_issues


def export_to_csv(issues: List[Dict], filename: str) -> None:
    """
    Export issues to a CSV file.
    
    Args:
        issues: List of issue dictionaries
        filename: Output CSV filename
        
    Raises:
        IOError: If file cannot be written
    """
    if not issues:
        print("No issues to export.")
        return
    
    fieldnames = [
        "identifier",
        "title",
        "completedAt",
        "creator_name",
        "assignee_name",
        "url",
        "team_name"
    ]
    
    with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        
        for issue in issues:
            row = {
                "identifier": issue.get("identifier", ""),
                "title": issue.get("title", ""),
                "completedAt": issue.get("completedAt", ""),
                "creator_name": safe_get_nested(issue, "creator", "name"),
                "assignee_name": safe_get_nested(issue, "assignee", "name"),
                "url": issue.get("url", ""),
                "team_name": safe_get_nested(issue, "team", "name")
            }
            writer.writerow(row)
    
    print(f"Successfully exported {len(issues)} issues to {filename}")


def main() -> None:
    """Main execution function."""
    # Get API key from environment (or .env file if python-dotenv is installed)
    api_key = os.environ.get("LINEAR_API_KEY")
    
    if not api_key:
        print("Error: LINEAR_API_KEY environment variable is not set.")
        print("\nOptions:")
        print("1. Set it directly: export LINEAR_API_KEY='your-api-key-here'")
        print("2. Create a .env file: cp env.template .env (then edit with your key)")
        sys.exit(1)
    
    if api_key == "your-api-key-here":
        print("Error: Please replace 'your-api-key-here' with your actual Linear API key.")
        sys.exit(1)
    
    print("Linear Issues Export - Last 12 Months")
    print("=" * 60)
    
    try:
        # Fetch all closed issues
        issues = fetch_closed_issues(api_key)
        
        # Export to CSV
        export_to_csv(issues, OUTPUT_FILENAME)
        
        print("=" * 60)
        print("Export completed successfully!")
        
    except requests.exceptions.RequestException as e:
        print(f"Network error: {e}", file=sys.stderr)
        sys.exit(1)
    except ValueError as e:
        print(f"Data error: {e}", file=sys.stderr)
        sys.exit(1)
    except IOError as e:
        print(f"File error: {e}", file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        print("\nOperation cancelled by user.")
        sys.exit(130)
    except Exception as e:
        print(f"Unexpected error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()