1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
|
#!/usr/bin/env python3
"""Seed a Linear workspace with Pearl's dogfooding conventions.
Creates, idempotently, via the Linear GraphQL API:
- a "Pearl" team whose workflow states are the six Agile columns
(icebox, triage queue, backlog, in-progress, done, cancelled),
- a "Pearl" project inside that team,
- three Custom Views: Pearl Open Issues, Pearl Icebox, Pearl Inbox.
Re-running is safe: anything already present by name is left alone.
The six columns are typed so pearl's own grouped view reproduces the board
order. icebox / triage queue / backlog are all the "backlog" Linear type with
ascending positions, so pearl's (state-type, position) group ordering lays them
out in that sequence; in-progress / done / cancelled take the started /
completed / canceled types.
Linear's CustomViewCreateInput has no grouping field, so "Pearl Open Issues"
is created with its filter only -- group it by category in pearl with
`pearl-set-grouping` (or in Linear's UI). Linear's teamCreate also seeds its
own default workflow states; this script adds the six Agile columns but does
not remove those defaults (tidy them in the Linear UI, or see a later
--tidy-default-states pass).
Usage:
LINEAR_API_KEY=lin_api_... python3 seed_pearl_workspace.py [--dry-run]
The key may also come from the LINEAR_API_KEY environment variable.
"""
import argparse
import json
import os
import sys
import urllib.request
from dataclasses import dataclass, field
LINEAR_GRAPHQL_URL = "https://api.linear.app/graphql"
TEAM_NAME = "Pearl"
TEAM_KEY = "PEARL"
PROJECT_NAME = "Pearl"
# The six Agile columns. icebox / triage queue / backlog share the "backlog"
# type with ascending positions so they order by position within one type;
# the rest take their natural Linear types. Colors are arbitrary but distinct.
TARGET_STATES = [
{"name": "icebox", "type": "backlog", "color": "#bec2c8", "position": 0},
{"name": "triage queue", "type": "backlog", "color": "#e2e2e2", "position": 1},
{"name": "backlog", "type": "backlog", "color": "#95a2b3", "position": 2},
{"name": "in-progress", "type": "started", "color": "#f2c94c", "position": 3},
{"name": "done", "type": "completed", "color": "#0f9d58", "position": 4},
{"name": "cancelled", "type": "canceled", "color": "#eb5757", "position": 5},
]
VIEW_NAMES = ["Pearl Open Issues", "Pearl Icebox", "Pearl Inbox"]
def desired_views(project_id):
"""Return the three view specs (name + IssueFilter filterData) for PROJECT_ID.
Pearl Open Issues scopes to the project and drops completed/canceled issues;
Pearl Icebox and Pearl Inbox scope to the project and a single column.
"""
proj = {"project": {"id": {"eq": project_id}}}
return [
{"name": "Pearl Open Issues",
"filterData": {**proj, "state": {"type": {"nin": ["completed", "canceled"]}}}},
{"name": "Pearl Icebox",
"filterData": {**proj, "state": {"name": {"eq": "icebox"}}}},
{"name": "Pearl Inbox",
"filterData": {**proj, "state": {"name": {"eq": "triage queue"}}}},
]
@dataclass
class Plan:
create_team: bool
create_states: list = field(default_factory=list)
create_project: bool = False
create_views: list = field(default_factory=list)
def plan_seed(current):
"""Given a workspace snapshot CURRENT, decide what to create.
CURRENT is {team, states, project, views}; team/project are dicts or None,
states/views are lists of dicts carrying at least a "name". Pure: no I/O.
"""
have_states = {s["name"] for s in current["states"]}
have_views = {v["name"] for v in current["views"]}
return Plan(
create_team=current["team"] is None,
create_states=[s for s in TARGET_STATES if s["name"] not in have_states],
create_project=current["project"] is None,
create_views=[{"name": n} for n in VIEW_NAMES if n not in have_views],
)
class LinearError(RuntimeError):
"""A GraphQL request returned an errors array."""
def _http_transport(api_key):
"""Return a transport(query, variables) -> response dict over real HTTP."""
def transport(query, variables):
body = json.dumps({"query": query, "variables": variables}).encode()
req = urllib.request.Request(
LINEAR_GRAPHQL_URL, data=body,
headers={"Authorization": api_key, "Content-Type": "application/json"})
with urllib.request.urlopen(req) as resp:
return json.loads(resp.read().decode())
return transport
class LinearClient:
"""Thin Linear GraphQL client; the only network boundary in this module."""
def __init__(self, api_key, transport=None):
self._transport = transport or _http_transport(api_key)
def execute(self, query, variables):
resp = self._transport(query, variables)
if resp.get("errors"):
msgs = "; ".join(e.get("message", str(e)) for e in resp["errors"])
raise LinearError(msgs)
return resp["data"]
def snapshot(self):
"""Read the current Pearl team, its states, the Pearl project, and views."""
teams = self.execute(
"query { teams { nodes { id key name "
"states { nodes { id name type position } } } } }", {})
team = next((t for t in teams["teams"]["nodes"] if t["name"] == TEAM_NAME), None)
states = team["states"]["nodes"] if team else []
projects = self.execute("query { projects { nodes { id name } } }", {})
project = next((p for p in projects["projects"]["nodes"]
if p["name"] == PROJECT_NAME), None)
views = self.execute("query { customViews { nodes { id name } } }", {})
view_nodes = [v for v in views["customViews"]["nodes"] if v["name"] in VIEW_NAMES]
return {"team": team, "states": states, "project": project, "views": view_nodes}
def create_team(self):
data = self.execute(
"mutation($input: TeamCreateInput!) { teamCreate(input: $input) "
"{ success team { id key name states { nodes { id name type } } } } }",
{"input": {"name": TEAM_NAME, "key": TEAM_KEY}})
return data["teamCreate"]["team"]
def create_state(self, team_id, spec):
data = self.execute(
"mutation($input: WorkflowStateCreateInput!) "
"{ workflowStateCreate(input: $input) { success workflowState { id name } } }",
{"input": {"teamId": team_id, "name": spec["name"], "type": spec["type"],
"color": spec["color"], "position": spec["position"]}})
return data["workflowStateCreate"]["workflowState"]
def create_project(self, team_id):
data = self.execute(
"mutation($input: ProjectCreateInput!) { projectCreate(input: $input) "
"{ success project { id name } } }",
{"input": {"name": PROJECT_NAME, "teamIds": [team_id]}})
return data["projectCreate"]["project"]
def create_view(self, team_id, spec):
data = self.execute(
"mutation($input: CustomViewCreateInput!) { customViewCreate(input: $input) "
"{ success customView { id name } } }",
{"input": {"name": spec["name"], "teamId": team_id, "shared": True,
"filterData": spec["filterData"]}})
return data["customViewCreate"]["customView"]
def seed(client):
"""Create the Pearl team, columns, project, and views; idempotent.
Returns a summary dict {team, project, states_created, views_created}.
"""
current = client.snapshot()
team = current["team"]
if team is None:
team = client.create_team()
current["states"] = team.get("states", {}).get("nodes", []) \
if isinstance(team.get("states"), dict) else (team.get("states") or [])
team_id = team["id"]
plan = plan_seed({**current, "team": team})
states_created = []
for spec in plan.create_states:
client.create_state(team_id, spec)
states_created.append(spec["name"])
project = current["project"] or client.create_project(team_id)
project_id = project["id"]
have_views = {v["name"] for v in current["views"]}
views_created = []
for spec in desired_views(project_id):
if spec["name"] not in have_views:
client.create_view(team_id, spec)
views_created.append(spec["name"])
# teamCreate seeds Linear's own default columns; report any team state that
# isn't one of the six so the user can tidy the board in the UI. Not
# auto-archived: a team's default state can't always be archived safely.
target_names = {s["name"] for s in TARGET_STATES}
leftover_states = [s["name"] for s in current["states"]
if s["name"] not in target_names]
return {"team": team_id, "project": project_id,
"states_created": states_created, "views_created": views_created,
"leftover_states": leftover_states}
def main(argv=None):
parser = argparse.ArgumentParser(description="Seed a Pearl Linear workspace.")
parser.add_argument("--dry-run", action="store_true",
help="print the plan without creating anything")
args = parser.parse_args(argv)
api_key = os.environ.get("LINEAR_API_KEY")
if not api_key:
print("LINEAR_API_KEY is not set", file=sys.stderr)
return 2
client = LinearClient(api_key)
if args.dry_run:
plan = plan_seed(client.snapshot())
print("Plan:")
print(f" create team: {plan.create_team}")
print(f" create states: {[s['name'] for s in plan.create_states]}")
print(f" create project: {plan.create_project}")
print(f" create views: {[v['name'] for v in plan.create_views]}")
return 0
summary = seed(client)
print(f"Seeded Pearl workspace: team {summary['team']}, project {summary['project']}")
print(f" states created: {summary['states_created'] or 'none'}")
print(f" views created: {summary['views_created'] or 'none'}")
if summary["leftover_states"]:
print(f" tidy in Linear UI (Linear's default columns, not removed): "
f"{summary['leftover_states']}")
return 0
if __name__ == "__main__":
sys.exit(main())
|
