misc: address new ruff requirements in python script

This commit is contained in:
Gaetan Lepage
2026-08-10 15:46:33 +00:00
committed by Gaétan Lepage
parent c2e6de075b
commit 9cdd979953
4 changed files with 55 additions and 47 deletions
+1 -2
View File
@@ -11,7 +11,6 @@ import json
import subprocess import subprocess
import sys import sys
from pathlib import Path from pathlib import Path
from typing import List
def run_nix_eval(file: str, *args: str) -> str: def run_nix_eval(file: str, *args: str) -> str:
@@ -30,7 +29,7 @@ def run_nix_eval(file: str, *args: str) -> str:
sys.exit() sys.exit()
def extract_maintainers(changed_files: List[str], pr_author: str) -> List[str]: def extract_maintainers(changed_files: list[str], pr_author: str) -> list[str]:
"""Extract maintainers from changed plugin files.""" """Extract maintainers from changed plugin files."""
if not changed_files: if not changed_files:
print("No plugin files changed. No maintainers to tag.", file=sys.stderr) print("No plugin files changed. No maintainers to tag.", file=sys.stderr)
+22 -24
View File
@@ -19,6 +19,8 @@ logging.basicConfig(
stream=sys.stderr, stream=sys.stderr,
) )
logger = logging.getLogger(__name__)
MANUAL_REVIEW_REQUEST_QUERY: Final[str] = """ MANUAL_REVIEW_REQUEST_QUERY: Final[str] = """
query($owner: String!, $repo: String!, $prNumber: Int!) { query($owner: String!, $repo: String!, $prNumber: Int!) {
repository(owner: $owner, name: $repo) { repository(owner: $owner, name: $repo) {
@@ -46,8 +48,6 @@ query($owner: String!, $repo: String!, $prNumber: Int!) {
class GHError(Exception): class GHError(Exception):
"""Custom exception for errors related to 'gh' CLI commands.""" """Custom exception for errors related to 'gh' CLI commands."""
pass
def run_gh_command( def run_gh_command(
args: list[str], args: list[str],
@@ -66,8 +66,8 @@ def run_gh_command(
) )
return result return result
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
logging.error("Error running command: %s", " ".join(command)) logger.error("Error running command: %s", " ".join(command))
logging.error("Stderr: %s", e.stderr.strip()) logger.error("Stderr: %s", e.stderr.strip())
raise GHError(f"Failed to execute gh command: {e}") from e raise GHError(f"Failed to execute gh command: {e}") from e
@@ -108,7 +108,7 @@ def get_manually_requested_reviewers(
} }
return manually_requested return manually_requested
except (GHError, json.JSONDecodeError, KeyError) as e: except (GHError, json.JSONDecodeError, KeyError) as e:
logging.error("Could not determine manually requested reviewers: %s", e) logger.error("Could not determine manually requested reviewers: %s", e)
return set() return set()
@@ -118,7 +118,7 @@ def get_users_from_gh(args: list[str], error_message: str) -> set[str]:
result = run_gh_command(args) result = run_gh_command(args)
return {user.strip() for user in result.stdout.split("\n") if user.strip()} return {user.strip() for user in result.stdout.split("\n") if user.strip()}
except GHError as e: except GHError as e:
logging.error("%s: %s", error_message, e) logger.error("%s: %s", error_message, e)
return set() return set()
@@ -164,13 +164,13 @@ def is_collaborator(owner: str, repo: str, username: str) -> bool:
return True return True
if "HTTP 404" in result.stderr: if "HTTP 404" in result.stderr:
logging.error("'%s' is not a collaborator in this repository.", username) logger.error("'%s' is not a collaborator in this repository.", username)
return False return False
else: else:
logging.error( logger.error(
"Unexpected error checking collaborator status for '%s'.", username "Unexpected error checking collaborator status for '%s'.", username
) )
logging.error("Stderr: %s", result.stderr.strip()) logger.error("Stderr: %s", result.stderr.strip())
raise GHError( raise GHError(
f"Unexpected API error for user '{username}': {result.stderr.strip()}" f"Unexpected API error for user '{username}': {result.stderr.strip()}"
) )
@@ -185,7 +185,7 @@ def update_reviewers(
) -> None: ) -> None:
"""Adds or removes reviewers from a PR in a single operation per action.""" """Adds or removes reviewers from a PR in a single operation per action."""
if reviewers_to_add: if reviewers_to_add:
logging.info("Requesting reviews from: %s", ", ".join(reviewers_to_add)) logger.info("Requesting reviews from: %s", ", ".join(reviewers_to_add))
try: try:
run_gh_command( run_gh_command(
[ [
@@ -197,12 +197,10 @@ def update_reviewers(
] ]
) )
except GHError as e: except GHError as e:
logging.error("Failed to add reviewers: %s", e) logger.error("Failed to add reviewers: %s", e)
if reviewers_to_remove and owner and repo: if reviewers_to_remove and owner and repo:
logging.info( logger.info("Removing review requests from: %s", ", ".join(reviewers_to_remove))
"Removing review requests from: %s", ", ".join(reviewers_to_remove)
)
payload = json.dumps({"reviewers": list(reviewers_to_remove)}) payload = json.dumps({"reviewers": list(reviewers_to_remove)})
try: try:
run_gh_command( run_gh_command(
@@ -217,7 +215,7 @@ def update_reviewers(
input_data=payload, input_data=payload,
) )
except GHError as e: except GHError as e:
logging.error("Failed to remove reviewers: %s", e) logger.error("Failed to remove reviewers: %s", e)
def main() -> None: def main() -> None:
@@ -256,20 +254,20 @@ def main() -> None:
args.owner, args.repo, args.pr_number, args.bot_user_name args.owner, args.repo, args.pr_number, args.bot_user_name
) )
logging.info("Current Maintainers: %s", " ".join(maintainers) or "None") logger.info("Current Maintainers: %s", " ".join(maintainers) or "None")
logging.info("Pending Reviewers: %s", " ".join(pending_reviewers) or "None") logger.info("Pending Reviewers: %s", " ".join(pending_reviewers) or "None")
logging.info("Past Reviewers: %s", " ".join(past_reviewers) or "None") logger.info("Past Reviewers: %s", " ".join(past_reviewers) or "None")
logging.info("Manually Requested: %s", " ".join(manually_requested) or "None") logger.info("Manually Requested: %s", " ".join(manually_requested) or "None")
# --- 2. Determine reviewers to remove --- # --- 2. Determine reviewers to remove ---
reviewers_to_remove: set[str] = set() reviewers_to_remove: set[str] = set()
if no_plugin_files: if no_plugin_files:
reviewers_to_remove = pending_reviewers - manually_requested reviewers_to_remove = pending_reviewers - manually_requested
logging.info("No plugin files changed. Removing bot-requested reviewers.") logger.info("No plugin files changed. Removing bot-requested reviewers.")
else: else:
outdated_reviewers = pending_reviewers - maintainers outdated_reviewers = pending_reviewers - maintainers
reviewers_to_remove = outdated_reviewers - manually_requested reviewers_to_remove = outdated_reviewers - manually_requested
logging.info("Removing outdated bot-requested reviewers.") logger.info("Removing outdated bot-requested reviewers.")
if reviewers_to_remove: if reviewers_to_remove:
update_reviewers( update_reviewers(
@@ -279,7 +277,7 @@ def main() -> None:
reviewers_to_remove=reviewers_to_remove, reviewers_to_remove=reviewers_to_remove,
) )
else: else:
logging.info("No reviewers to remove.") logger.info("No reviewers to remove.")
# --- 3. Determine new reviewers to add --- # --- 3. Determine new reviewers to add ---
reviewers_to_add: set[str] = set() reviewers_to_add: set[str] = set()
@@ -295,14 +293,14 @@ def main() -> None:
non_collaborators = potential_reviewers - reviewers_to_add non_collaborators = potential_reviewers - reviewers_to_add
if non_collaborators: if non_collaborators:
logging.warning( logger.warning(
"Ignoring non-collaborators: %s", ", ".join(non_collaborators) "Ignoring non-collaborators: %s", ", ".join(non_collaborators)
) )
if reviewers_to_add: if reviewers_to_add:
update_reviewers(args.pr_number, reviewers_to_add=reviewers_to_add) update_reviewers(args.pr_number, reviewers_to_add=reviewers_to_add)
else: else:
logging.info("No new reviewers to add.") logger.info("No new reviewers to add.")
if __name__ == "__main__": if __name__ == "__main__":
@@ -72,23 +72,27 @@ class MetaMaintainerGenerator:
capture_output=True, capture_output=True,
text=True, text=True,
timeout=60, timeout=60,
check=True,
) )
if result.returncode == 0:
data = json.loads(result.stdout)
print("✅ Successfully extracted maintainers using meta.maintainers")
return data
else:
print(f"❌ Failed to extract maintainers: {result.stderr}")
sys.exit(1)
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
print("❌ Timeout while extracting maintainers") print("❌ Timeout while extracting maintainers")
sys.exit(1) sys.exit(1)
except Exception as e: except subprocess.CalledProcessError as e:
print(f"❌ Failed to extract maintainers: {e.stderr}")
sys.exit(1)
except OSError as e:
print(f"❌ Error extracting maintainers: {e}") print(f"❌ Error extracting maintainers: {e}")
sys.exit(1) sys.exit(1)
try:
data = json.loads(result.stdout)
except json.JSONDecodeError as e:
print(f"❌ Error parsing extracted maintainers: {e}")
sys.exit(1)
print("✅ Successfully extracted maintainers using meta.maintainers")
return data
def generate_maintainers_file(self) -> None: def generate_maintainers_file(self) -> None:
"""Generate the complete all-maintainers.nix file.""" """Generate the complete all-maintainers.nix file."""
print("📄 Generating all-maintainers.nix using meta.maintainers...") print("📄 Generating all-maintainers.nix using meta.maintainers...")
@@ -132,24 +136,24 @@ class MetaMaintainerGenerator:
def validate_generated_file(self) -> bool: def validate_generated_file(self) -> bool:
"""Validate the generated Nix file syntax.""" """Validate the generated Nix file syntax."""
try: try:
result = subprocess.run( subprocess.run(
["nix-instantiate", "--eval", str(self.output_file), "--strict"], ["nix-instantiate", "--eval", str(self.output_file), "--strict"],
capture_output=True, capture_output=True,
text=True, text=True,
timeout=10, timeout=10,
check=True,
) )
except subprocess.CalledProcessError as e:
if result.returncode == 0: print("❌ Warning: Generated file has Nix syntax errors")
print("✅ Generated file has valid Nix syntax") print(e.stderr[:500])
return True return False
else: except (OSError, subprocess.TimeoutExpired) as e:
print("❌ Warning: Generated file has Nix syntax errors")
print(result.stderr[:500])
return False
except Exception as e:
print(f"Warning: Could not validate file: {e}") print(f"Warning: Could not validate file: {e}")
return False return False
print("✅ Generated file has valid Nix syntax")
return True
def print_statistics(self, maintainer_data: dict) -> None: def print_statistics(self, maintainer_data: dict) -> None:
"""Print generation statistics.""" """Print generation statistics."""
stats = maintainer_data["stats"] stats = maintainer_data["stats"]
@@ -203,7 +207,7 @@ def main():
except KeyboardInterrupt: except KeyboardInterrupt:
print("\n❌ Generation cancelled by user") print("\n❌ Generation cancelled by user")
sys.exit(1) sys.exit(1)
except Exception as e: except (OSError, KeyError) as e:
print(f"❌ Error generating maintainers file: {e}") print(f"❌ Error generating maintainers file: {e}")
sys.exit(1) sys.exit(1)
+7
View File
@@ -0,0 +1,7 @@
[lint]
ignore = [
# Some of our scripts are run through `nix-shell` as an interpreter, which
# requires a second `#!nix-shell ...` line right after the shebang.
# See https://nix.dev/manual/nix/latest/command-ref/nix-shell#use-as-a-interpreter
"EXE005",
]