Skip to content

Commit e2ce36b

Browse files
committed
migrate: Add ruleset helpers to the template
Copy the reusable GitHub ruleset helper functions to the migration script template so future migration steps can query and update rulesets without duplicating API plumbing. Signed-off-by: Leandro Lucarella <luca-frequenz@llucax.com>
1 parent 2c0d680 commit e2ce36b

1 file changed

Lines changed: 127 additions & 1 deletion

File tree

.github/cookiecutter-migrate.template.py

Lines changed: 127 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828
import sys
2929
import tempfile
3030
from pathlib import Path
31-
from typing import SupportsIndex
31+
from typing import Any, SupportsIndex
3232

3333
_manual_steps: list[str] = [] # pylint: disable=invalid-name
3434

@@ -180,6 +180,132 @@ def calculate_file_sha256_skip_lines(filepath: Path, skip_lines: int) -> str | N
180180
return hashlib.sha256(remaining_content.encode()).hexdigest()
181181

182182

183+
def find_ruleset(name: str) -> dict[str, Any] | None:
184+
"""Find a repository ruleset by name using the GitHub API.
185+
186+
Args:
187+
name: The name of the ruleset to search for.
188+
189+
Returns:
190+
The ruleset summary dict (id, name, …) if found, or ``None`` if not
191+
found or if the API call failed (a diagnostic is printed in the latter
192+
case).
193+
"""
194+
try:
195+
stdout = subprocess.check_output(
196+
["gh", "api", "repos/:owner/:repo/rulesets"],
197+
text=True,
198+
stderr=subprocess.PIPE,
199+
)
200+
except FileNotFoundError:
201+
print(" gh CLI not found; cannot query rulesets via the GitHub API.")
202+
return None
203+
except subprocess.CalledProcessError as exc:
204+
print(f" Failed to list rulesets: {exc.stderr.strip()}")
205+
return None
206+
207+
rulesets: list[dict[str, Any]] = json.loads(stdout)
208+
return next((r for r in rulesets if r.get("name") == name), None)
209+
210+
211+
def get_ruleset(ruleset: str | int) -> dict[str, Any] | None:
212+
"""Fetch the full details of a repository ruleset by name or ID.
213+
214+
Args:
215+
ruleset: The ruleset name (``str``) or numeric ruleset ID (``int``).
216+
217+
Returns:
218+
The full ruleset dict, or ``None`` if the ruleset could not be found
219+
or the API call failed (a diagnostic is printed).
220+
"""
221+
ruleset_id = ruleset
222+
if isinstance(ruleset, str):
223+
entry = find_ruleset(ruleset)
224+
if entry is None:
225+
return None
226+
ruleset_id = entry["id"]
227+
228+
try:
229+
stdout = subprocess.check_output(
230+
["gh", "api", f"repos/:owner/:repo/rulesets/{ruleset_id}"],
231+
text=True,
232+
stderr=subprocess.PIPE,
233+
)
234+
except subprocess.CalledProcessError as exc:
235+
print(f" Failed to fetch ruleset {ruleset_id}: {exc.stderr.strip()}")
236+
return None
237+
238+
return json.loads(stdout) # type: ignore[no-any-return]
239+
240+
241+
def update_ruleset(ruleset_id: int, config: dict[str, Any]) -> bool:
242+
"""Update a repository ruleset via the GitHub API.
243+
244+
Only ``name``, ``target``, ``enforcement``, ``conditions``, ``rules``,
245+
and ``bypass_actors`` are sent (explicit allowlist to avoid sending
246+
read-only fields back to the API).
247+
248+
Args:
249+
ruleset_id: The numeric ruleset ID to update.
250+
config: The full ruleset dict (as returned by :func:`get_ruleset`)
251+
with the desired changes already applied in-memory.
252+
253+
Returns:
254+
``True`` on success, ``False`` if the API call failed (a diagnostic
255+
is printed).
256+
"""
257+
payload: dict[str, Any] = {
258+
"name": config["name"],
259+
"target": config["target"],
260+
"enforcement": config["enforcement"],
261+
"conditions": config["conditions"],
262+
"rules": config["rules"],
263+
}
264+
if "bypass_actors" in config:
265+
payload["bypass_actors"] = config["bypass_actors"]
266+
267+
try:
268+
subprocess.check_output(
269+
[
270+
"gh",
271+
"api",
272+
"-X",
273+
"PUT",
274+
f"repos/:owner/:repo/rulesets/{ruleset_id}",
275+
"--input",
276+
"-",
277+
],
278+
input=json.dumps(payload),
279+
text=True,
280+
stderr=subprocess.PIPE,
281+
)
282+
except subprocess.CalledProcessError as exc:
283+
print(f" Failed to update ruleset {ruleset_id}: {exc.stderr.strip()}")
284+
return False
285+
286+
return True
287+
288+
289+
def get_ruleset_settings_url() -> str | None:
290+
"""Return the URL to the repository's ruleset settings page.
291+
292+
Returns:
293+
The URL as a string, or ``None`` if it could not be determined.
294+
"""
295+
try:
296+
stdout = subprocess.check_output(
297+
["gh", "repo", "view", "--json", "owner,name"],
298+
text=True,
299+
stderr=subprocess.PIPE,
300+
)
301+
info: dict[str, Any] = json.loads(stdout)
302+
org = info["owner"]["login"]
303+
repo = info["name"]
304+
return f"https://github.com/{org}/{repo}/settings/rules"
305+
except (subprocess.CalledProcessError, KeyError, json.JSONDecodeError):
306+
return None
307+
308+
183309
def manual_step(message: str) -> None:
184310
"""Print a manual step message in yellow."""
185311
_manual_steps.append(message)

0 commit comments

Comments
 (0)