Original: https://github.com/sparkmeter/sentry2csv (modified for User-Agent extraction)

sentry2csvUA.py

#!/usr/bin/env python3
"""Export a Sentry project's issues to CSV."""

import argparse
import asyncio
import csv
import logging
import sys
from typing import Any, Dict, List, Optional, Tuple, Union

import aiohttp

logging.basicConfig()
logger = logging.getLogger(__name__)


class Sentry2CSVException(Exception):
    """A handled exception."""

    def __init__(self, message):  # pylint: disable=super-init-not-called
        self.message = message


async def fetch(
        session: aiohttp.ClientSession, url: str, params=None
) -> Tuple[Union[List[Dict[str, Any]], Dict[str, Any]], Dict[str, Dict[str, str]]]:
    """Fetch JSON from a URL."""
    logger.debug("Fetching %s with params: %s", url, params)
    async with session.get(url, params=params) as response:
        logger.debug("Received response: %s", response)
        if response.status == 403:
            raise Sentry2CSVException(f"Failed to query Sentry: access denied.")
        return await response.json(), response.links


async def enrich_issue(
        session: aiohttp.ClientSession, issue: Dict[str, Any]
) -> None:
    """Enrich an issue with data from the latest event."""
    event, _ = await fetch(session, f'https://sentry.io/api/0/issues/{issue["id"]}/events/latest/')
    issue["_enrichments"] = {}

    issue["_enrichments"]['IP'] = event['user']['ip_address']
    issue["_enrichments"]['Received'] = event['dateReceived']
    for e in event['entries']:
        if e['type'] == 'request':
            for [k, v] in e['data']['headers']:
                if k == 'User-Agent':
                    issue["_enrichments"]['UserAgent'] = v


async def fetch_issues(session: aiohttp.ClientSession, issues_url: str) -> List[Dict[str, Any]]:
    """Fetch all issues from Sentry."""
    page_count = 1
    issues: List[Dict[str, Any]] = []
    cursor = ""
    while True:
        print(f"Fetching issues page {page_count}")
        resp, links = await fetch(
            session, issues_url, params={"cursor": cursor, "statsPeriod": "", "query": "is:unresolved"}
        )
        logger.debug("Received page %s", resp)
        assert isinstance(resp, list), f"Bad response type. Expected list, got {type(resp)}"
        issues.extend(resp)
        if links.get("next", {}).get("results") != "true":
            break
        cursor = links["next"]["cursor"]
        page_count += 1
    return issues


def write_csv(filename: str, issues: List[Dict[str, Any]]):
    """Write Sentry issues to CSV."""
    fieldnames = ["Error", "Location", "Details", "Events", "Users", "Notes", "Link"]
    if issues and "_enrichments" in issues[0]:
        fieldnames.extend(issues[0]["_enrichments"].keys())
    with open(filename, "w") as outfile:
        writer = csv.DictWriter(outfile, fieldnames=fieldnames)
        writer.writeheader()
        for issue in issues:
            try:
                # mapping from
                #  https://github.com/getsentry/sentry/blob/9910cc917d2def63b110e75d4d17dedf7f415f58/src/sentry/static/sentry/app/utils/events.tsx#L7  # pylint: disable=line-too-long
                issue_type = issue["type"]
                if issue_type == "error":
                    error = issue["metadata"].get("type", issue_type)  # get more specific if we can
                    details = issue["metadata"]["value"]
                elif issue_type == "csp":
                    error = "csp"
                    details = issue["metadata"]["message"]
                elif issue_type == "default":
                    error = "default"
                    details = issue["metadata"].get("title", "")
                else:
                    logger.debug("Unknown issue type: %s\n%s", issue_type, issue)
                    error = issue_type
                    details = ""
                row = {
                    "Error": error,
                    "Location": issue["culprit"],
                    "Details": details,
                    "Events": issue["count"],
                    "Users": issue["userCount"],
                    "Notes": "",
                    "Link": issue["permalink"],
                }
                row = {**row, **issue.get("_enrichments", {})}
                writer.writerow(row)
            except KeyError as kerr:
                logger.debug("Failed to process row, missing key: %s\n%s", kerr, issue)
                raise Sentry2CSVException(f"Unexpected API response. Run with -vv to debug.") from kerr


async def export(token: str, organization: str, project: str):
    """Export data from Sentry to CSV."""
    issues_url = f"https://sentry.io/api/0/projects/{organization}/{project}/issues/"
    async with aiohttp.ClientSession(headers={"Authorization": f"Bearer {token}"}) as session:
        try:
            issues = await fetch_issues(session, issues_url)
            print(f"Enriching {len(issues)} issues with event data...")
            await asyncio.gather(
                *[asyncio.ensure_future(enrich_issue(session, issue)) for issue in issues]
            )

            outfile = f"{organization}-{project}-export.csv"
            write_csv(outfile, issues)
            print(f"Exported to {outfile}")
        except Sentry2CSVException as err:
            print(f"Export failed. {err.message}")
            sys.exit(1)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description="Export a Sentry project's issues to CSV")
    parser.add_argument("-v", "--verbose", default=0, action="count", help="Increase the log verbosity.")
    parser.add_argument("--token", metavar="API_TOKEN", nargs=1, required=True, help="The Sentry API token")
    parser.add_argument("organization", metavar="ORGANIZATION", nargs=1, help="The Sentry organization")
    parser.add_argument("project", metavar="PROJECT", nargs=1, help="The Sentry project")
    args = parser.parse_args()
    if args.verbose > 1:
        logger.setLevel(logging.DEBUG)
    elif args.verbose == 1:
        logger.setLevel(logging.INFO)
    else:
        logger.setLevel(logging.WARNING)
    loop = asyncio.get_event_loop()
    loop.run_until_complete(export(args.token[0], args.organization[0], args.project[0]))