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