chore(lint): run formatter on python files

Author: dmcilvaney

.github/workflows/scripts/components/compute_render_set.py

@@ -74,10 +74,12 @@ def main() -> None:
     # changed list AND with hand-edited specs would otherwise print twice,
     # and a component with N modified spec files would print N times.
     # dict.fromkeys preserves first-seen order.
-    names = dict.fromkeys([
-        *from_changed(entries),
-        *from_specs_diff(args.specs_diff_file, args.specs_dir, renderable),
-    ])
+    names = dict.fromkeys(
+        [
+            *from_changed(entries),
+            *from_specs_diff(args.specs_diff_file, args.specs_dir, renderable),
+        ]
+    )
     for name in names:
         print(name)
 

.github/workflows/scripts/control-tower/client.py

@@ -28,9 +28,7 @@
 # (azl-ControlTower/ControlTower/Shared/Models/Jobs/JobStatus.cs).
 NON_TERMINAL_STATUSES = frozenset({"Queued", "Pending", "Running"})
 SUCCESS_STATUS = "Completed"
-TERMINAL_FAILURE_STATUSES = frozenset(
-    {"Failed", "Cancelled", "CancelledByAdmin", "Unknown", "TimedOut"}
-)
+TERMINAL_FAILURE_STATUSES = frozenset({"Failed", "Cancelled", "CancelledByAdmin", "Unknown", "TimedOut"})
 
 
 @dataclass
@@ -86,9 +84,7 @@ def format_error(response: requests.Response) -> str:
       * ASP.NET validation: ``{"title", "errors": {field: [msg, ...]}}``
     """
     method = response.request.method if response.request is not None else "?"
-    lines: list[str] = [
-        f"HTTP {response.status_code} {response.reason} from {method} {response.url}"
-    ]
+    lines: list[str] = [f"HTTP {response.status_code} {response.reason} from {method} {response.url}"]
 
     body: Any
     try:
@@ -178,8 +174,7 @@ def _parse_json_object(response: requests.Response, context: str) -> dict:
         body = response.json()
     except ValueError as exc:
         raise RuntimeError(
-            f"{context} returned HTTP {response.status_code} "
-            f"but the body was not valid JSON:\n{response.text}"
+            f"{context} returned HTTP {response.status_code} but the body was not valid JSON:\n{response.text}"
         ) from exc
     if not isinstance(body, dict):
         raise RuntimeError(
@@ -219,9 +214,7 @@ def post_scenario(
         json_payload=payload,
     )
     if not response.ok:
-        raise RuntimeError(
-            f"Control Tower '{context}' request failed.\n" + format_error(response)
-        )
+        raise RuntimeError(f"Control Tower '{context}' request failed.\n" + format_error(response))
     return _parse_json_object(response, f"Control Tower '{context}'")
 
 
@@ -235,13 +228,9 @@ def get_job_status(
 ) -> dict:
     """GET the job status. Refreshes the bearer token on 401 and retries once."""
     url = f"{base_url}/api/Workflow/jobs/status/{job_id}"
-    response = _request_with_refresh(
-        session, "GET", url, credential, audience, token_holder
-    )
+    response = _request_with_refresh(session, "GET", url, credential, audience, token_holder)
     if not response.ok:
-        raise RuntimeError(
-            "Control Tower job status request failed.\n" + format_error(response)
-        )
+        raise RuntimeError("Control Tower job status request failed.\n" + format_error(response))
     return _parse_json_object(response, "Control Tower job status")
 
 
@@ -284,19 +273,13 @@ def poll_until_terminal(
     job_status_object: dict = {}
 
     while True:
-        job_status_object = get_job_status(
-            session, base_url, credential, audience, token_holder, job_id
-        )
+        job_status_object = get_job_status(session, base_url, credential, audience, token_holder, job_id)
         current_status = job_status_object.get("status", "Unknown")
         elapsed = int(time.monotonic() - start)
 
         if current_status != previous_status:
             task_summary = _summarize_tasks(job_status_object.get("tasks"))
-            transition = (
-                f"{previous_status} -> {current_status}"
-                if previous_status is not None
-                else current_status
-            )
+            transition = f"{previous_status} -> {current_status}" if previous_status is not None else current_status
             suffix = f" | {task_summary}" if task_summary else ""
             print(
                 f"Job {job_id} status: {transition} (elapsed {elapsed}s){suffix}",
@@ -342,14 +325,7 @@ def report_failure(final: dict) -> None:
 
     tasks = final.get("tasks")
     if isinstance(tasks, list):
-        failed = [
-            t
-            for t in tasks
-            if isinstance(t, dict) and t.get("status") in TERMINAL_FAILURE_STATUSES
-        ]
+        failed = [t for t in tasks if isinstance(t, dict) and t.get("status") in TERMINAL_FAILURE_STATUSES]
         for task in failed:
             name = task.get("taskName") or task.get("taskId")
-            print(
-                f"##[error]task '{name}' status={task.get('status')} "
-                f"attempt={task.get('attemptNumber')}"
-            )
+            print(f"##[error]task '{name}' status={task.get('status')} attempt={task.get('attemptNumber')}")

.github/workflows/scripts/control-tower/run_package_build.py

@@ -39,16 +39,12 @@ def _load_build_components(path: Path) -> list[str]:
     try:
         raw = path.read_text(encoding="utf-8")
     except OSError as exc:
-        raise SystemExit(
-            f"##[error]Failed to read --changed-components-file {path!s}: {exc}"
-        ) from exc
+        raise SystemExit(f"##[error]Failed to read --changed-components-file {path!s}: {exc}") from exc
 
     try:
         entries = json.loads(raw)
     except json.JSONDecodeError as exc:
-        raise SystemExit(
-            f"##[error]--changed-components-file {path!s} is not valid JSON: {exc}"
-        ) from exc
+        raise SystemExit(f"##[error]--changed-components-file {path!s} is not valid JSON: {exc}") from exc
 
     if not isinstance(entries, list):
         raise SystemExit(
@@ -212,10 +208,7 @@ def main() -> None:
 
     job_id = build_response.get("jobId")
     if not job_id:
-        print(
-            "##[error]Control Tower 'package' response did not include a 'jobId'. "
-            "Cannot confirm job acceptance."
-        )
+        print("##[error]Control Tower 'package' response did not include a 'jobId'. Cannot confirm job acceptance.")
         sys.exit(1)
 
     # ── Brief poll — just confirm the job was accepted ───────────────

.github/workflows/scripts/control-tower/run_prcheck.py

@@ -49,16 +49,12 @@ def _load_components_from_file(path: Path) -> list[str]:
     try:
         raw = path.read_text(encoding="utf-8")
     except OSError as exc:
-        raise SystemExit(
-            f"##[error]Failed to read --changed-components-file {path!s}: {exc}"
-        ) from exc
+        raise SystemExit(f"##[error]Failed to read --changed-components-file {path!s}: {exc}") from exc
 
     try:
         entries = json.loads(raw)
     except json.JSONDecodeError as exc:
-        raise SystemExit(
-            f"##[error]--changed-components-file {path!s} is not valid JSON: {exc}"
-        ) from exc
+        raise SystemExit(f"##[error]--changed-components-file {path!s} is not valid JSON: {exc}") from exc
 
     if not isinstance(entries, list):
         raise SystemExit(
@@ -89,9 +85,7 @@ def _parse_args() -> argparse.Namespace:
         required=True,
         help="Entra ID audience URI (e.g. api://<client-id>)",
     )
-    parser.add_argument(
-        "--api-base-url", required=True, help="Base URL of the Control Tower service"
-    )
+    parser.add_argument("--api-base-url", required=True, help="Base URL of the Control Tower service")
     parser.add_argument(
         "--build-reason",
         required=True,
@@ -182,16 +176,11 @@ def main() -> None:
     print(json.dumps(payload, indent=2))
 
     if args.build_reason == "PullRequest":
-        print(
-            "Skipping Control Tower call - pull request triggers are not supported, yet."
-        )
+        print("Skipping Control Tower call - pull request triggers are not supported, yet.")
         return
 
     if not components:
-        print(
-            "No affected components detected between source and target commits; "
-            "skipping Control Tower call."
-        )
+        print("No affected components detected between source and target commits; skipping Control Tower call.")
         return
 
     # ── Acquire bearer token ─────────────────────────────────────────
@@ -221,17 +210,11 @@ def main() -> None:
 
     job_id = prcheck_response.get("jobId")
     if not job_id:
-        print(
-            "##[error]Control Tower 'prcheck' response did not include a 'jobId'. "
-            "Cannot poll for job status."
-        )
+        print("##[error]Control Tower 'prcheck' response did not include a 'jobId'. Cannot poll for job status.")
         sys.exit(1)
 
     # ── Poll for job completion ──────────────────────────────────────
-    print(
-        f"Polling job {job_id} every {args.poll_interval_seconds}s "
-        f"(timeout {args.poll_timeout_seconds}s)..."
-    )
+    print(f"Polling job {job_id} every {args.poll_interval_seconds}s (timeout {args.poll_timeout_seconds}s)...")
     try:
         final, timed_out = ct.poll_until_terminal(
             session,

.github/workflows/scripts/locks-check/post_locks_comment.py

@@ -117,19 +117,12 @@ def parse_update_output(path: Path) -> list[dict]:
 
     if not isinstance(data, list):
         raise SystemExit(
-            f"Error: update output has unexpected shape (expected null or list, "
-            f"got {type(data).__name__})"
+            f"Error: update output has unexpected shape (expected null or list, got {type(data).__name__})"
         )
 
     for entry in data:
-        if (
-            not isinstance(entry, dict)
-            or "component" not in entry
-            or "changed" not in entry
-        ):
-            raise SystemExit(
-                f"Error: update output entry has unexpected shape: {entry!r}"
-            )
+        if not isinstance(entry, dict) or "component" not in entry or "changed" not in entry:
+            raise SystemExit(f"Error: update output entry has unexpected shape: {entry!r}")
 
     return [entry for entry in data if entry["changed"] is True]
 
@@ -160,9 +153,7 @@ def format_comment(
     # inject arbitrary commands into a maintainer's terminal. Fall back to
     # `-a` if any name fails the same regex used for display so the
     # printed command is always safe to run as-is.
-    use_all = n_changed > 30 or any(
-        not _SAFE_NAME_RE.match(name) for name in comp_names
-    )
+    use_all = n_changed > 30 or any(not _SAFE_NAME_RE.match(name) for name in comp_names)
     remediation_cmd = _update_command([] if use_all else comp_names, use_all=use_all)
 
     lines: list[str] = [
@@ -248,9 +239,7 @@ def format_comment(
 
 
 def _gh(*args: str) -> str:
-    return subprocess.run(
-        ["gh", *args], capture_output=True, text=True, check=True
-    ).stdout.strip()
+    return subprocess.run(["gh", *args], capture_output=True, text=True, check=True).stdout.strip()
 
 
 def find_existing_comments(repo: str, pr: str) -> list[str]:
@@ -267,11 +256,7 @@ def find_existing_comments(repo: str, pr: str) -> list[str]:
             "--paginate",
             f"/repos/{repo}/issues/{pr}/comments",
             "--jq",
-            (
-                f'.[] | select(.user.login == "{BOT_AUTHOR}") '
-                f'| select(.body | contains("{COMMENT_MARKER}")) '
-                "| .id"
-            ),
+            (f'.[] | select(.user.login == "{BOT_AUTHOR}") | select(.body | contains("{COMMENT_MARKER}")) | .id'),
         )
     except subprocess.CalledProcessError:
         return []
@@ -342,9 +327,7 @@ def delete_comment_if_exists(repo: str, pr: str) -> None:
 
 
 def main() -> int:
-    parser = argparse.ArgumentParser(
-        description="Post `azldev component update` drift as a PR comment."
-    )
+    parser = argparse.ArgumentParser(description="Post `azldev component update` drift as a PR comment.")
     parser.add_argument(
         "--update-output",
         type=Path,
@@ -353,9 +336,7 @@ def main() -> int:
     )
     parser.add_argument("--repo", required=True, help="GitHub repo (owner/repo)")
     parser.add_argument("--pr", required=True, help="PR number")
-    parser.add_argument(
-        "--artifacts-url", default=None, help="Direct URL to patch artifact"
-    )
+    parser.add_argument("--artifacts-url", default=None, help="Direct URL to patch artifact")
     parser.add_argument("--run-id", default=None, help="GitHub Actions run ID")
     args = parser.parse_args()