Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222#!/usr/bin/env python3"""Format repository sources or check that they are already formatted."""
import argparseimport osimport shleximport subprocessimport sysfrom concurrent.futures import ThreadPoolExecutor, as_completedfrom dataclasses import dataclassfrom pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
@dataclass(frozen=True)class Command: args: tuple[str, ...] cwd: Path = REPO_ROOT
@dataclass(frozen=True)class FormatterGroup: name: str commands: tuple[Command, ...]
@dataclass(frozen=True)class FormatterResult: name: str output: str returncode: int
def just_formatter_group(*, check: bool) -> FormatterGroup: args = ["just", "--unstable", "--fmt"] if check: args.append("--check") return FormatterGroup("Just", (Command(tuple(args)),))
def rust_formatter_group(*, check: bool) -> FormatterGroup: rust_workspace = REPO_ROOT / "codex-rs" repository_files = subprocess.check_output( [ "git", "ls-files", "-z", "--cached", "--others", "--exclude-standard", "--", "*.rs", ], cwd=REPO_ROOT, ).split(b"\0") rust_files = sorted( os.path.relpath(REPO_ROOT / os.fsdecode(encoded_path), rust_workspace) for encoded_path in repository_files if encoded_path and (REPO_ROOT / os.fsdecode(encoded_path)).is_file() ) # Git supplies every source file, including files Cargo cannot discover. # Do not recursively format child modules again or include ignored files. args = [ "rustfmt", "--edition", "2024", "--config-path", str(rust_workspace / "rustfmt.toml"), "--config", "imports_granularity=Item,skip_children=true", ] if check: args.append("--check") # Batch relative paths to stay below Windows command-line limits. commands = tuple( Command(tuple(args + rust_files[start : start + 64]), rust_workspace) for start in range(0, len(rust_files), 64) ) return FormatterGroup("Rust", commands)
def buildifier_formatter_group(*, check: bool) -> FormatterGroup: repository_files = subprocess.check_output( ["git", "ls-files", "-z", "--cached", "--others", "--exclude-standard"], cwd=REPO_ROOT, ).split(b"\0") buildifier_files: list[str] = [] for encoded_path in repository_files: if not encoded_path: continue path = Path(os.fsdecode(encoded_path)) name = path.name if ( name in {"BUILD", "WORKSPACE", "MODULE.bazel"} or name.startswith(("BUILD.", "WORKSPACE.")) or name.endswith((".BUILD.bazel", ".MODULE.bazel", ".bzl", ".sky")) or ".bzl." in name or ".sky." in name ): buildifier_files.append(path.as_posix()) buildifier_files.sort()
# Invoke DotSlash explicitly because Windows does not honor shebangs. buildifier_args = [ "dotslash", str(REPO_ROOT / "tools" / "buildifier"), "-mode=check" if check else "-mode=fix", "-lint=off", *buildifier_files, ] return FormatterGroup("Bazel/Starlark", (Command(tuple(buildifier_args)),))
def python_sdk_formatter_group(*, check: bool) -> FormatterGroup: # Each `--project` retains its local dependency and Ruff configuration context. uv_run_args = [ "uv", "run", "--frozen", "--project", "sdk/python", "--only-group", "format", ] format_args = [ *uv_run_args, "ruff", "format", ] if check: format_args.append("--check") # `ruff check --diff` reports lint-driven rewrites without changing files. # It is the check-mode counterpart of `--fix --fix-only`, not a full lint gate. lint_args = ["ruff", "check", "--diff"] else: # Ruff's lint fixer and formatter are separate passes: the first applies # fixable lint rewrites, while the second formats source layout. lint_args = ["ruff", "check", "--fix", "--fix-only"]
return FormatterGroup( "Python SDK", ( Command((*uv_run_args, *lint_args, "sdk/python")), Command((*format_args, "sdk/python")), ), )
def python_scripts_formatter_group(*, check: bool) -> FormatterGroup: args = ["uv", "run", "--frozen", "--project", "scripts", "ruff", "format"] if check: args.append("--check") args.append(".") return FormatterGroup("Python scripts", (Command(tuple(args)),))
def formatter_groups(*, check: bool) -> tuple[FormatterGroup, ...]: return ( just_formatter_group(check=check), rust_formatter_group(check=check), buildifier_formatter_group(check=check), python_sdk_formatter_group(check=check), python_scripts_formatter_group(check=check), )
def run_formatter_group(group: FormatterGroup) -> FormatterResult: """Run one formatter group sequentially and return its buffered output.""" for command in group.commands: try: process = subprocess.run( command.args, cwd=command.cwd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, ) except OSError as error: output = f"$ {shlex.join(command.args)}\n{error}\n" return FormatterResult(group.name, output, 1)
if process.returncode != 0: output = f"$ {shlex.join(command.args)}\n{process.stdout}" if process.stdout and not process.stdout.endswith("\n"): output += "\n" return FormatterResult(group.name, output, process.returncode)
return FormatterResult(group.name, "", 0)
def main() -> int: parser = argparse.ArgumentParser() parser.add_argument( "--check", action="store_true", help="check formatting without modifying files", ) args = parser.parse_args() groups = formatter_groups(check=args.check)
failures: list[str] = [] with ThreadPoolExecutor(max_workers=len(groups)) as executor: futures = [executor.submit(run_formatter_group, group) for group in groups] for future in as_completed(futures): result = future.result() if result.returncode != 0: failures.append(result.name) print(f"==> {result.name} formatter failed", file=sys.stderr) print(result.output, end="", file=sys.stderr)
if failures: print(f"Formatting failed: {', '.join(failures)}", file=sys.stderr) return 1 return 0
if __name__ == "__main__": raise SystemExit(main())