Skip to content

Commit d9cc4bc

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 2e7e8ff commit d9cc4bc

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
@@ -27,7 +27,7 @@
2727
import sys
2828
import tempfile
2929
from pathlib import Path
30-
from typing import SupportsIndex
30+
from typing import Any, SupportsIndex
3131

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

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

181181

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

0 commit comments

Comments
 (0)