misc.python.materialize.linear

  1# Copyright Materialize, Inc. and contributors. All rights reserved.
  2#
  3# Use of this software is governed by the Business Source License
  4# included in the LICENSE file at the root of this repository.
  5#
  6# As of the Change Date specified in that file, in accordance with
  7# the Business Source License, use of this software will be governed
  8# by the Apache License, Version 2.0.
  9
 10"""Linear utilities."""
 11
 12import os
 13import re
 14from typing import Any
 15
 16import requests
 17
 18from materialize.github import (
 19    CI_APPLY_TO,
 20    CI_IGNORE_FAILURE,
 21    CI_LOCATION,
 22    CI_RE,
 23    GitHubIssueWithInvalidRegexp,
 24    KnownGitHubIssue,
 25)
 26
 27LINEAR_CLOSED_STATE_TYPES = {"completed", "canceled", "duplicate"}
 28LINEAR_STALE_STATE_NAMES = {"stale"}
 29
 30
 31def _search_issues_graphql(token: str) -> list[dict[str, Any]]:
 32    query = """
 33    query($cursor: String) {
 34      issues(
 35        filter: { description: { contains: "ci-regexp:" } }
 36        first: 100
 37        after: $cursor
 38        includeArchived: false
 39      ) {
 40        nodes {
 41          identifier
 42          title
 43          description
 44          url
 45          state {
 46            type
 47          }
 48        }
 49        pageInfo {
 50          hasNextPage
 51          endCursor
 52        }
 53      }
 54    }
 55    """
 56
 57    all_issues: list[dict[str, Any]] = []
 58    cursor = None
 59
 60    while True:
 61        variables: dict[str, Any] = {}
 62        if cursor:
 63            variables["cursor"] = cursor
 64
 65        response = requests.post(
 66            "https://api.linear.app/graphql",
 67            headers={
 68                "Authorization": token,
 69                "Content-Type": "application/json",
 70            },
 71            json={"query": query, "variables": variables},
 72            timeout=60,
 73        )
 74
 75        if response.status_code != 200:
 76            raise ValueError(
 77                f"Bad return code from Linear GraphQL: {response.status_code}, "
 78                f"response={response.text[:500]}, "
 79                f"has_token=True"
 80            )
 81
 82        result = response.json()
 83        if "errors" in result:
 84            raise ValueError(f"Linear GraphQL errors: {result['errors']}")
 85
 86        search_data = result["data"]["issues"]
 87        for node in search_data["nodes"]:
 88            if node is None:
 89                continue
 90            description = node.get("description") or ""
 91            if "ci-regexp:" in description:
 92                all_issues.append(node)
 93
 94        if not search_data["pageInfo"]["hasNextPage"]:
 95            break
 96        cursor = search_data["pageInfo"]["endCursor"]
 97
 98    return all_issues
 99
100
101def get_known_issues_from_linear(
102    token: str | None = os.getenv("LINEAR_READ_ONLY_TOKEN"),
103) -> tuple[list[KnownGitHubIssue], list[GitHubIssueWithInvalidRegexp]]:
104    if not token:
105        return ([], [])
106
107    issues = _search_issues_graphql(token)
108
109    known_issues = []
110    issues_with_invalid_regex = []
111
112    for issue in issues:
113        body = issue.get("description") or ""
114
115        state_type = issue.get("state", {}).get("type", "")
116        state = "CLOSED" if state_type in LINEAR_CLOSED_STATE_TYPES else "OPEN"
117
118        info = {
119            "number": issue["identifier"],
120            "title": issue["title"],
121            "body": body,
122            "url": issue["url"],
123            "state": state,
124            "source": "linear",
125        }
126
127        matches = CI_RE.findall(body)
128        matches_apply_to = CI_APPLY_TO.findall(body)
129        matches_location = CI_LOCATION.findall(body)
130        matches_ignore_failure = CI_IGNORE_FAILURE.findall(body)
131
132        if len(matches) > 1:
133            issues_with_invalid_regex.append(
134                GitHubIssueWithInvalidRegexp(
135                    internal_error_type="LINEAR_INVALID_REGEXP",
136                    issue_url=issue["url"],
137                    issue_title=issue["title"],
138                    issue_number=issue["identifier"],
139                    regex_pattern=f"Multiple regexes, but only one supported: {[match.strip() for match in matches]}",
140                )
141            )
142            continue
143
144        if len(matches_ignore_failure) > 1:
145            issues_with_invalid_regex.append(
146                GitHubIssueWithInvalidRegexp(
147                    internal_error_type="LINEAR_INVALID_IGNORE_FAILURE",
148                    issue_url=issue["url"],
149                    issue_title=issue["title"],
150                    issue_number=issue["identifier"],
151                    regex_pattern=f"Multiple ci-ignore-failures, but only one supported: {[match.strip() for match in matches_ignore_failure]}",
152                )
153            )
154            continue
155
156        if len(matches) == 0:
157            continue
158
159        if len(matches_location) >= 2:
160            issues_with_invalid_regex.append(
161                GitHubIssueWithInvalidRegexp(
162                    internal_error_type="LINEAR_INVALID_LOCATION",
163                    issue_url=issue["url"],
164                    issue_title=issue["title"],
165                    issue_number=issue["identifier"],
166                    regex_pattern=f"Multiple ci-locations, but only one supported: {[match.strip() for match in matches_location]}",
167                )
168            )
169            continue
170
171        location: str | None = (
172            matches_location[0] if len(matches_location) == 1 else None
173        )
174
175        ignore_failure = len(matches_ignore_failure) == 1 and matches_ignore_failure[
176            0
177        ].strip() in ("true", "yes", "1")
178
179        try:
180            regex_pattern = re.compile(matches[0].strip().encode())
181        except:
182            issues_with_invalid_regex.append(
183                GitHubIssueWithInvalidRegexp(
184                    internal_error_type="LINEAR_INVALID_REGEXP",
185                    issue_url=issue["url"],
186                    issue_title=issue["title"],
187                    issue_number=issue["identifier"],
188                    regex_pattern=matches[0].strip(),
189                )
190            )
191            continue
192
193        if matches_apply_to:
194            for match_apply_to in matches_apply_to:
195                known_issues.append(
196                    KnownGitHubIssue(
197                        regex_pattern,
198                        match_apply_to.strip().lower(),
199                        info,
200                        ignore_failure,
201                        location,
202                    )
203                )
204        else:
205            known_issues.append(
206                KnownGitHubIssue(regex_pattern, None, info, ignore_failure, location)
207            )
208
209    return (known_issues, issues_with_invalid_regex)