Skip to content

Commit 6c18a32

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 8c903d6 commit 6c18a32

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
@@ -29,7 +29,7 @@
2929
import sys
3030
import tempfile
3131
from pathlib import Path
32-
from typing import SupportsIndex
32+
from typing import Any, SupportsIndex
3333

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

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

183183

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

0 commit comments

Comments
 (0)