diff --git a/github_scripts/suite_data.py b/github_scripts/suite_data.py index a5dbc367..3d23a27d 100644 --- a/github_scripts/suite_data.py +++ b/github_scripts/suite_data.py @@ -8,14 +8,17 @@ Class containing helper methods for gathering data needed for a SuiteReport object """ -import re +import ast import os +import re import sqlite3 import subprocess -import yaml from collections import defaultdict from pathlib import Path -from typing import Dict, List, Optional, Set, Union +from typing import Optional, Union + +import yaml + from git_bdiff import GitBDiff, GitInfo @@ -41,7 +44,7 @@ def __init__(self, suite_path: Path = None) -> None: self.source_root = suite_path / "share" / "source" self.task_states = {} - def get_um_failed_configs(self) -> Set[str]: + def get_um_failed_configs(self) -> set[str]: """ Read through failed UM rose_ana tasks """ @@ -74,7 +77,7 @@ def read_um_section(self, change: str) -> str: section = "Unknown" return section - def get_changed_um_section(self) -> Set[str]: + def get_changed_um_section(self) -> set[str]: """ Read through bdiff of UM source and find code owner section for each changed file @@ -111,7 +114,7 @@ def get_changed_um_section(self) -> Set[str]: return changed_sections - def get_um_owners(self, filename: str) -> Dict: + def get_um_owners(self, filename: str) -> dict: """ Read UM Code Owners file and write to a dictionary """ @@ -149,7 +152,7 @@ def get_um_owners(self, filename: str) -> Dict: return owners - def parse_tasks(self) -> Dict[str, List[str]]: + def parse_tasks(self) -> dict[str, list[str]]: """ Read through the tasks run, sorting by state """ @@ -217,7 +220,7 @@ def determine_primary_source(self) -> str: return "unknown" - def read_rose_conf(self) -> Dict[str, str]: + def read_rose_conf(self) -> dict[str, str]: """ Read the suite rose-suite.conf file into a dictionary """ @@ -271,7 +274,7 @@ def find_unknown_dependency(self, dependency: str) -> str: return match.group(1).rstrip("/") raise RuntimeError(f"Unable to find source for dependency {dependency}") - def read_dependencies(self) -> Dict[str, Dict]: + def read_dependencies(self) -> dict[str, dict]: """ Read the suite dependencies from the dependencies.yaml file - this is assumed to have been copied to the suite_path directory @@ -328,7 +331,7 @@ def get_suite_starttime(self) -> str: starttime = row[0] return starttime.split("+")[0] - def read_groups_run(self) -> List[str]: + def read_groups_run(self) -> list[str]: """ Read in groups run as part of suite from the cylc database file """ @@ -338,13 +341,15 @@ def read_groups_run(self) -> List[str]: self.suite_path / "log" / "db", ["key", "value"], "workflow_template_vars" ): if row[0] in ("g", "group"): - groups = row[1].strip("[]'\"").split(",") + groups = ast.literal_eval(row[1]) + # optionally strip whitespaces from elements + groups = [item.strip() for item in groups] break if groups is None: groups = ["suite_default"] return groups - def get_task_states(self) -> Dict[str, str]: + def get_task_states(self) -> dict[str, str]: """ Query the database and return a dictionary of states. This is assumed to be in suite_path/log/db @@ -358,8 +363,8 @@ def get_task_states(self) -> Dict[str, str]: return data def query_suite_database( - self, database: Path, selections: List[str], source: str - ) -> List[tuple]: + self, database: Path, selections: list[str], source: str + ) -> list[tuple]: """ Create an sql statement and query provided database. Return the result """ @@ -376,7 +381,7 @@ def query_suite_database( return data def run_command( - self, command: Union[str, List[str]], shell: bool = False, rval: bool = False + self, command: Union[str, list[str]], shell: bool = False, rval: bool = False ) -> Optional[subprocess.CompletedProcess]: """ Run a subprocess command and return the result object diff --git a/github_scripts/suite_report_git.py b/github_scripts/suite_report_git.py index 62dc2b8e..a9acafa8 100755 --- a/github_scripts/suite_report_git.py +++ b/github_scripts/suite_report_git.py @@ -16,12 +16,11 @@ from collections import defaultdict from contextlib import contextmanager from pathlib import Path -from typing import Dict, List, Set, Tuple from suite_data import SuiteData -def create_markdown_row(*columns: str, header=False) -> List[str]: +def create_markdown_row(*columns: str, header=False) -> list[str]: """ Join any number of columns into a markdown formatted table row Will attempt to format column entries as str @@ -73,16 +72,16 @@ def __init__(self, suite_path: Path) -> None: self.suite_starttime: str = self.get_suite_starttime() self.workflow_id: str = self.get_workflow_id() self.cylc_url: str = self.generate_cylc_url() - self.task_states: Dict[str, str] = self.get_task_states() - self.groups: List[str] = self.read_groups_run() - self.rose_data: Dict[str, str] = self.read_rose_conf() - self.dependencies: Dict[str, Dict] = self.read_dependencies() + self.task_states: dict[str, str] = self.get_task_states() + self.groups: list[str] = self.read_groups_run() + self.rose_data: dict[str, str] = self.read_rose_conf() + self.dependencies: dict[str, dict] = self.read_dependencies() self.primary_source: str = self.determine_primary_source() self.populate_gitinfo() self.populate_gitbdiff() self.trac_log = [] - def parse_local_source(self, source: str) -> Tuple[str, str]: + def parse_local_source(self, source: str) -> tuple[str, str]: """ Find the branch name or hash and remote reference for a given source """ @@ -162,7 +161,7 @@ def create_dependency_table(self) -> None: self.trac_log.append("") - def create_task_tables(self, parsed_tasks: Dict[str, List[str]]) -> None: + def create_task_tables(self, parsed_tasks: dict[str, list[str]]) -> None: """ Create tables containing summary of task states and number of tasks won """ @@ -200,12 +199,12 @@ def create_task_tables(self, parsed_tasks: Dict[str, List[str]]) -> None: self.trac_log.extend(create_markdown_row(task, state)) self.trac_log.append(self.close_collapsed) - def create_um_code_owner_table(self, owners: Dict) -> None: + def create_um_code_owner_table(self, owners: dict) -> None: """ Create a table of required UM code owner approvals """ - changed_sections: Set[str] = self.get_changed_um_section() + changed_sections: set[str] = self.get_changed_um_section() if changed_sections: self.trac_log.extend( create_markdown_row("Section", "Owner", "Deputy", "State", header=True) @@ -218,12 +217,12 @@ def create_um_code_owner_table(self, owners: Dict) -> None: else: self.trac_log.append("* No UM Code Owners Required") - def create_um_config_owner_table(self, owners: Dict) -> None: + def create_um_config_owner_table(self, owners: dict) -> None: """ Create a table of required UM config owner approvals """ - failed_configs: Set[str] = self.get_um_failed_configs() + failed_configs: set[str] = self.get_um_failed_configs() if not failed_configs: self.trac_log.append("No UM Config Owners Required") return