#!/usr/bin/env vpython3
#
# /// script
# requires-python = '>=3.11,<3.12'
# dependencies = [
# 'ruff==0.15.17',
# 'yapf==0.40.2',
# 'platformdirs==4.1.0',
# 'importlib-metadata==7.0.0',
# 'tomli==2.0.1',
# 'zipp==3.7.0'
# ]
# ///
# Copyright 2026 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import os
import sys
import argparse
import subprocess
import json
import difflib
import tomllib
import tempfile
from typing import NamedTuple
if "RUFF_CACHE_DIR" not in os.environ:
os.environ["RUFF_CACHE_DIR"] = os.path.join(
tempfile.gettempdir(), ".ruff_cache"
)
KNOWN_SUBCOMMANDS = {
"check",
"format",
"linter",
"rule",
"config",
"explain",
"clean",
"timezone",
"version",
"help",
"server",
"analyze",
}
class LineRange(NamedTuple):
"""Represents a 1-based line range interval [start, end) where end is exclusive."""
start: int
end: int
class ParsedArguments(NamedTuple):
"""Holds the parsed line ranges and the scrubbed pass-through arguments."""
ranges: list[LineRange]
pass_through_args: list[str]
class FormattingOptions(NamedTuple):
"""Isolates the parsed formatting options and positional target files."""
has_format: bool
has_diff: bool
has_check: bool
subcommand: str | None
target_files: list[str]
chain_base_args: list[str]
subcmd_idx: int
_dir_config_cache: dict[tuple[str, str | None], str | None] = {}
def resolve_config_for_dir(
target_dir: str, root_dir: str | None = None
) -> str | None:
search_dir = os.path.realpath(os.path.abspath(target_dir))
abs_root = os.path.realpath(os.path.abspath(root_dir)) if root_dir else None
cache_key = (search_dir, abs_root)
traversed_dirs = []
while True:
if cache_key in _dir_config_cache:
resolved = _dir_config_cache[cache_key]
for d in traversed_dirs:
_dir_config_cache[(d, abs_root)] = resolved
return resolved
traversed_dirs.append(search_dir)
# 1. Check .style.yapf first (prioritize YAPF)
if os.path.isfile(os.path.join(search_dir, ".style.yapf")):
resolved = "yapf"
break
# 2. Check ruff.toml / .ruff.toml
if os.path.isfile(
os.path.join(search_dir, ".ruff.toml")
) or os.path.isfile(os.path.join(search_dir, "ruff.toml")):
resolved = "ruff"
break
# 3. Check pyproject.toml
pyproject = os.path.join(search_dir, "pyproject.toml")
if os.path.isfile(pyproject):
try:
with open(pyproject, "rb") as f:
cfg = tomllib.loads(
f.read().decode("utf-8", errors="ignore")
)
tool = cfg.get("tool", {})
has_ruff = "ruff" in tool
has_yapf = "yapf" in tool
has_other = "black" in tool or "pyink" in tool
if has_ruff and not (has_yapf or has_other):
resolved = "ruff"
break
elif has_yapf:
# If both Ruff and YAPF are configured, we prioritize YAPF.
# We stop traversal here (closest config wins) rather than
# falling back to a potential Ruff config higher in the tree.
resolved = "yapf"
break
elif has_other:
# If another unsupported formatter (like Black) is configured,
# we must NOT format it with Ruff, and we shouldn't fall back
# to a parent YAPF config either.
resolved = None
break
except Exception as e:
sys.stderr.write(f"Warning: failed to parse {pyproject}: {e}\n")
if abs_root and search_dir == abs_root:
resolved = None
break
parent_dir = os.path.realpath(os.path.dirname(search_dir))
if parent_dir == search_dir:
resolved = None
break
search_dir = parent_dir
cache_key = (search_dir, abs_root)
for d in traversed_dirs:
_dir_config_cache[(d, abs_root)] = resolved
return resolved
_ruff_bin = None
def get_ruff_bin() -> str:
global _ruff_bin
if _ruff_bin is None:
import ruff
_ruff_bin = ruff.find_ruff_bin()
return _ruff_bin
def extract_root_flag(args: list[str]) -> tuple[str | None, list[str]]:
"""Extracts --root, --top-dir, or --root-dir from args using argparse."""
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument(
"--root", "--top-dir", "--root-dir", dest="root_dir", default=None
)
opts, remaining = parser.parse_known_args(args)
root_dir = os.path.abspath(opts.root_dir) if opts.root_dir else None
return root_dir, remaining
def should_use_ruff(target_path: str, root_dir: str | None = None) -> bool:
"""Determines whether to format target_path using Ruff or fall back to YAPF.
Args:
target_path: Path to the file or directory being formatted.
Returns:
True if covered by a valid Ruff configuration, False otherwise (fallback to YAPF).
"""
abs_target = os.path.abspath(target_path)
search_dir = (
os.path.dirname(abs_target)
if (os.path.isfile(abs_target) or not os.path.exists(abs_target))
else abs_target
)
return resolve_config_for_dir(search_dir, root_dir) == "ruff"
def has_yapf_config(target_path: str, root_dir: str | None = None) -> bool:
"""Determines whether a YAPF formatting configuration governs target_path.
Args:
target_path: Path to the file or directory being evaluated.
Returns:
True if .style.yapf or pyproject.toml with [tool.yapf] exists in any
ancestor directory, False otherwise.
"""
abs_target = os.path.abspath(target_path)
search_dir = (
os.path.dirname(abs_target)
if (os.path.isfile(abs_target) or not os.path.exists(abs_target))
else abs_target
)
return resolve_config_for_dir(search_dir, root_dir) == "yapf"
def parse_range(s: str) -> LineRange:
"""Parses a range string 'startLine:startCol-endLine:endCol' or 'startLine-endLine'.
Args:
s: Range argument string.
Returns:
A LineRange instance containing 1-based start and end line numbers.
Raises:
ValueError: If range format is invalid or start/end line numbers are invalid.
"""
parts = s.split("-")
if len(parts) != 2:
raise ValueError(f"invalid range format (missing '-'): {s!r}")
start_parts = parts[0].split(":")
end_parts = parts[1].split(":")
try:
start_line = int(start_parts[0])
except ValueError as e:
raise ValueError(f"invalid start line: {e}") from e
if start_line < 1:
raise ValueError(f"start line {start_line} must be >= 1")
if len(start_parts) > 1:
try:
int(start_parts[1]) # Validate start col is int
except ValueError as e:
raise ValueError(f"invalid start column: {e}") from e
try:
end_line = int(end_parts[0])
except ValueError as e:
raise ValueError(f"invalid end line: {e}") from e
has_end_col = len(end_parts) > 1
if has_end_col:
try:
end_col = int(end_parts[1])
except ValueError as e:
raise ValueError(f"invalid end column: {e}") from e
else:
end_col = 1
if start_line > end_line:
raise ValueError(
f"start line {start_line} is greater than end line {end_line}"
)
# If end column is not specified, we assume the range is inclusive of end_line.
# If end column is specified and > 1, it includes parts of the end line.
# In both cases, we increment end_line to convert to exclusive [start, end) range.
if not has_end_col or end_col > 1:
end_line += 1
return LineRange(start=start_line, end=end_line)
def merge_ranges(ranges: list[LineRange]) -> list[LineRange]:
"""Merges overlapping or adjacent line ranges.
Assumes the input list is already sorted ascending by start line.
Merging adjacent ranges (e.g. [1, 5) and [5, 10)) prevents redundant formatting calls.
Args:
ranges: List of LineRange items sorted ascending by start line.
Returns:
List of merged LineRange items sorted ascending.
"""
if len(ranges) <= 1:
return list(ranges)
merged: list[LineRange] = []
curr_start, curr_end = ranges[0].start, ranges[0].end
for r in ranges[1:]:
if r.start <= curr_end:
curr_end = max(curr_end, r.end)
else:
merged.append(LineRange(start=curr_start, end=curr_end))
curr_start, curr_end = r.start, r.end
merged.append(LineRange(start=curr_start, end=curr_end))
return merged
def parse_ranges(args: list[str]) -> ParsedArguments:
"""Extracts --range flags and gathers pass-through arguments using argparse."""
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument("--range", action="append", dest="ranges", default=[])
opts, pass_through_args = parser.parse_known_args(args)
ranges: list[LineRange] = []
for r in opts.ranges:
try:
ranges.append(parse_range(r))
except ValueError as e:
raise ValueError(f"invalid --range value {r!r}: {e}") from e
return ParsedArguments(ranges=ranges, pass_through_args=pass_through_args)
def parse_formatting_options(pass_through_args: list[str]) -> FormattingOptions:
"""Scans and isolates formatting subcommands, diff/check flags, target files, and index.
Args:
pass_through_args: List of command-line arguments after extracting range flags.
Returns:
A FormattingOptions dataclass instance containing the parsed configuration.
"""
has_format = False
has_diff = False
has_check = False
subcommand: str | None = None
target_files: list[str] = []
chain_base_args: list[str] = []
flags_with_values = {
"--config",
"-c",
"--target-version",
"--line-length",
"--stdin-filename",
"--exclude",
"--extension",
}
skip_next = False
subcmd_idx = -1
for arg in pass_through_args:
if skip_next:
skip_next = False
chain_base_args.append(arg)
continue
if arg in KNOWN_SUBCOMMANDS and subcommand is None:
subcommand = arg
if arg == "format":
has_format = True
subcmd_idx = len(chain_base_args)
chain_base_args.append(arg)
elif arg == "--diff":
has_diff = True
elif arg == "--check":
has_check = True
elif arg in ("-i", "--in-place"):
pass
elif arg in flags_with_values:
chain_base_args.append(arg)
skip_next = True
elif not arg.startswith("-") and arg != "-":
target_files.append(arg)
if len(target_files) > 1:
chain_base_args.append(arg)
else:
chain_base_args.append(arg)
return FormattingOptions(
has_format=has_format,
has_diff=has_diff,
has_check=has_check,
subcommand=subcommand,
target_files=target_files,
chain_base_args=chain_base_args,
subcmd_idx=subcmd_idx,
)
def execute_ranges_chain(
chain_base_args: list[str],
subcmd_idx: int,
ranges: list[LineRange],
original_val: bytes,
) -> tuple[int, bytes]:
"""Runs ruff sequentially over in-memory buffers for all merged ranges.
Args:
chain_base_args: Base arguments for intermediate runs.
subcmd_idx: The index of the subcommand 'format' in chain_base_args.
ranges: The list of parsed LineRange items.
original_val: The original byte value read from disk.
Returns:
A tuple of (exit_code, formatted_val).
"""
current_val = original_val
ranges.sort(key=lambda r: r.start)
merged = merge_ranges(ranges)
merged.reverse()
for r in merged:
range_arg = f"--range={r.start}:1-{r.end}:1"
if subcmd_idx != -1:
step_args = (
chain_base_args[: subcmd_idx + 1]
+ [range_arg]
+ chain_base_args[subcmd_idx + 1 :]
)
else:
step_args = [range_arg] + chain_base_args
cmd = [get_ruff_bin()] + step_args
proc = subprocess.run(cmd, input=current_val, capture_output=True)
if proc.returncode != 0:
sys.stderr.buffer.write(proc.stderr)
sys.stderr.write(f"ruff failed on range {r.start}-{r.end}\n")
return proc.returncode, b""
current_val = proc.stdout
return 0, current_val
def dispatch_output(
target: str,
original_val: bytes,
current_val: bytes,
has_check: bool,
has_diff: bool,
) -> int:
"""Emits diffs to stdout, returns checks, or writes formatted contents back to disk.
Args:
target: Target filepath.
original_val: The original byte contents of target.
current_val: The formatted byte contents.
has_check: True if '--check' was specified.
has_diff: True if '--diff' was specified.
Returns:
Exit code (0 on success, non-zero on error or check failure).
"""
if has_diff:
orig_lines = original_val.decode("utf-8", errors="replace").splitlines(
keepends=True
)
curr_lines = current_val.decode("utf-8", errors="replace").splitlines(
keepends=True
)
diff_lines = difflib.unified_diff(
orig_lines, curr_lines, fromfile=target, tofile=target
)
diff_bytes = "".join(diff_lines).encode("utf-8")
sys.stdout.buffer.write(diff_bytes)
sys.stdout.buffer.flush()
if has_check:
return 1 if current_val != original_val else 0
if has_diff:
return 0
if current_val != original_val:
try:
with open(target, "wb") as f:
f.write(current_val)
except Exception as e:
sys.stderr.write(f"failed to write {target}: {e}\n")
return 1
return 0
def run_ruff_with_ranges(args: list[str]) -> int:
"""Formats multiple regions of a target file by chaining Ruff over in-memory stdin/stdout.
Args:
args: Command-line arguments passed to the ruff wrapper.
Returns:
Exit code (0 on success, non-zero on error or check failure).
"""
ruff_args = list(args)
try:
parsed_args = parse_ranges(ruff_args)
except ValueError as e:
sys.stderr.write(f"{e}\n")
return 1
opts = parse_formatting_options(parsed_args.pass_through_args)
if opts.subcommand is None:
ruff_args.insert(0, "format")
try:
parsed_args = parse_ranges(ruff_args)
except ValueError as e:
sys.stderr.write(f"{e}\n")
return 1
opts = parse_formatting_options(parsed_args.pass_through_args)
# We are calling ruff as is if there is only one or no ranges.
if len(parsed_args.ranges) < 2:
cmd = [get_ruff_bin()] + ruff_args + ["--force-exclude"]
return subprocess.call(cmd)
if not opts.has_format:
sys.stderr.write(
"multiple ranges are only supported for the 'format' subcommand\n"
)
return 1
is_stdin = "-" in parsed_args.pass_through_args
if not is_stdin and len(opts.target_files) != 1:
sys.stderr.write(
"multi-range formatting requires either stdin '-' or exactly one target file on disk\n"
)
return 1
if is_stdin:
original_val = sys.stdin.buffer.read()
target = opts.target_files[0] if opts.target_files else "stdin"
else:
target = opts.target_files[0]
try:
with open(target, "rb") as f:
original_val = f.read()
except Exception as e:
sys.stderr.write(f"failed to read {target}: {e}\n")
return 1
has_stdin_filename = any(
a == "--stdin-filename" or a.startswith("--stdin-filename=")
for a in opts.chain_base_args
)
if not has_stdin_filename and not is_stdin:
opts.chain_base_args.append(f"--stdin-filename={target}")
if "--force-exclude" not in opts.chain_base_args:
opts.chain_base_args.append("--force-exclude")
if not is_stdin:
opts.chain_base_args.append("-")
exit_code, formatted_val = execute_ranges_chain(
opts.chain_base_args, opts.subcmd_idx, parsed_args.ranges, original_val
)
if exit_code != 0:
return exit_code
if is_stdin:
if opts.has_diff:
return dispatch_output(
target, original_val, formatted_val, opts.has_check, opts.has_diff
)
if opts.has_check:
return 1 if formatted_val != original_val else 0
sys.stdout.buffer.write(formatted_val)
sys.stdout.buffer.flush()
return 0
return dispatch_output(
target, original_val, formatted_val, opts.has_check, opts.has_diff
)
def translate_args(args: list[str]) -> list[str]:
"""Translates Ruff command-line arguments into YAPF command-line arguments.
Args:
args: List of command-line arguments passed to the ruff wrapper.
Returns:
List of translated arguments suitable for invoking YAPF.
"""
yapf_args = []
has_diff = False
has_stdin = False
files = []
i = 0
while i < len(args):
arg = args[i]
# YAPF takes options and files directly without a subcommand, so strip "format".
if arg == "format":
i += 1
continue
# Ruff uses --range start:col-end:col; YAPF uses --line start-end (line numbers only, no columns).
elif arg == "--range":
val = args[i + 1]
try:
r = parse_range(val)
yapf_start = r.start
yapf_end = max(yapf_start, r.end - 1)
yapf_args.extend(["--line", f"{yapf_start}-{yapf_end}"])
except Exception:
yapf_args.extend(["--range", val])
i += 2
# Handle equals-separated --range=start:col-end:col (e.g. from fmtserver or editors).
elif arg.startswith("--range="):
val = arg.split("=", 1)[1]
try:
r = parse_range(val)
yapf_start = r.start
yapf_end = max(yapf_start, r.end - 1)
yapf_args.extend(["--line", f"{yapf_start}-{yapf_end}"])
except Exception:
yapf_args.append(arg)
i += 1
elif arg == "--diff":
has_diff = True
yapf_args.append(arg)
i += 1
elif arg == "-":
has_stdin = True
i += 1
elif arg.startswith("--stdin-filename="):
i += 1
elif arg == "--stdin-filename":
i += 2
elif arg.startswith("-"):
# Pass through other flags
yapf_args.append(arg)
i += 1
else:
files.append(arg)
yapf_args.append(arg)
i += 1
# Ruff formats files in-place by default; YAPF prints to stdout unless -i is explicitly passed.
if files and not has_diff and not has_stdin:
yapf_args.append("-i")
return yapf_args
def get_yapf_style(target_path: str, root_dir: str | None = None) -> str:
abs_target = os.path.abspath(target_path)
search_dir = (
os.path.dirname(abs_target)
if (os.path.isfile(abs_target) or not os.path.exists(abs_target))
else abs_target
)
while True:
style_file = os.path.join(search_dir, ".style.yapf")
if os.path.isfile(style_file):
return style_file
pyproject = os.path.join(search_dir, "pyproject.toml")
if os.path.isfile(pyproject):
try:
with open(pyproject, "rb") as f:
content = f.read().decode("utf-8", errors="ignore")
if "[tool.yapf]" in content:
return pyproject
except Exception:
pass
if root_dir and os.path.realpath(
os.path.abspath(search_dir)
) == os.path.realpath(os.path.abspath(root_dir)):
break
parent_dir = os.path.realpath(os.path.dirname(search_dir))
if parent_dir == search_dir:
break
search_dir = parent_dir
return "pep8"
def format_ruff_batch_no_range(
paths: list[str], diff: bool, dry_run: bool
) -> int:
if not paths:
return 0
# Chunk paths on Windows to avoid the command line length limit (8191 characters).
# On other platforms, we run a single invocation for better performance.
chunk_size = 50 if sys.platform == "win32" else len(paths)
has_changes = False
has_error = False
for i in range(0, len(paths), chunk_size):
chunk = paths[i : i + chunk_size]
cmd = [get_ruff_bin(), "format", "--force-exclude"]
if diff:
cmd.append("--diff")
elif dry_run:
cmd.append("--check")
cmd.extend(chunk)
proc = subprocess.run(cmd, capture_output=True)
if proc.returncode > 1 or proc.returncode < 0:
sys.stderr.buffer.write(proc.stderr)
has_error = True
elif proc.returncode == 1 and dry_run:
has_changes = True
if diff or dry_run:
if diff:
sys.stdout.buffer.write(proc.stdout)
if has_error:
return 1
if has_changes:
return 2
return 0
def format_yapf_batch(
files: list[tuple[str, list[LineRange]]],
root_dir: str | None,
diff: bool,
dry_run: bool,
full: bool,
) -> int:
# The import of yapf is 77 ms performance hit, so call it only when needed.
import yapf
has_changes = False
has_error = False
for path, ranges in files:
if ranges and not full:
yapf_lines = [
(r.start, r.end - 1) for r in ranges if r.start < r.end
]
if not yapf_lines:
continue
else:
yapf_lines = None
in_place = not (diff or dry_run)
print_diff = diff
quiet = dry_run and not diff
style_config = get_yapf_style(path, root_dir)
try:
changed = yapf.FormatFiles(
[path],
lines=yapf_lines,
style_config=style_config,
in_place=in_place,
print_diff=print_diff,
quiet=quiet,
)
if changed and dry_run:
has_changes = True
except Exception as e:
sys.stderr.write(f"YAPF failed on {path}: {e}\n")
has_error = True
if has_error:
return 1
if has_changes:
return 2
return 0
def run_batch() -> int:
"""Runs formatting in batch mode by reading configuration from stdin.
The input JSON is expected to be a dictionary with the following schema:
{
"root": string (optional, root directory of the repository),
"diff": boolean (optional, whether to print diff instead of in-place format),
"dry_run": boolean (optional, same as diff for Ruff, prevents write for YAPF),
"full": boolean (optional, whether to format full files instead of ranges),
"files": [
{
"path": string (relative path to file),
"ranges": [[start, end], ...] (optional, 1-based line ranges.
NOTE: Ranges must use exclusive end lines [start, end),
matching Ruff's range semantics. The caller must convert
inclusive ranges by adding 1 to the end line.)
},
...
]
}
Returns:
Exit code (0 on success, non-zero on error).
"""
try:
config = json.load(sys.stdin)
except Exception as e:
sys.stderr.write(f"Failed to parse batch config JSON: {e}\n")
return 1
if not isinstance(config, dict):
sys.stderr.write("Batch config must be a JSON object\n")
return 1
root_dir = config.get("root")
if root_dir is not None:
if not isinstance(root_dir, str):
sys.stderr.write("Batch config 'root' must be a string\n")
return 1
root_dir = os.path.abspath(root_dir)
files_cfg = config.get("files", [])
if not isinstance(files_cfg, list):
sys.stderr.write("Batch config 'files' must be a list\n")
return 1
for f in files_cfg:
if (
not isinstance(f, dict)
or "path" not in f
or not isinstance(f["path"], str)
):
sys.stderr.write("Invalid file entry in batch config\n")
return 1
ranges_cfg = f.get("ranges", [])
if not isinstance(ranges_cfg, list):
sys.stderr.write("File 'ranges' must be a list\n")
return 1
for r in ranges_cfg:
if (
not isinstance(r, list)
or len(r) != 2
or not isinstance(r[0], int)
or not isinstance(r[1], int)
):
sys.stderr.write("Invalid range entry in batch config\n")
return 1
diff = config.get("diff", False)
dry_run = config.get("dry_run", False)
full = config.get("full", False)
ruff_files_no_range = []
yapf_files = []
for f in files_cfg:
path = f["path"]
if root_dir and not os.path.isabs(path):
path = os.path.join(root_dir, path)
ranges = [LineRange(r[0], r[1]) for r in f.get("ranges", [])]
if should_use_ruff(path, root_dir):
# Ruff files are formatted in their entirety to maximize batching
# performance. This is safe because onboarded files are already
# fully formatted, so formatting the whole file won't affect
# untouched lines.
ruff_files_no_range.append(path)
elif has_yapf_config(path, root_dir):
yapf_files.append((path, ranges))
has_changes = False
has_error = False
if ruff_files_no_range:
code = format_ruff_batch_no_range(ruff_files_no_range, diff, dry_run)
if code == 1:
has_error = True
elif code == 2:
has_changes = True
if yapf_files:
code = format_yapf_batch(yapf_files, root_dir, diff, dry_run, full)
if code == 1:
has_error = True
elif code == 2:
has_changes = True
if has_error:
return 1
if has_changes:
return 2
return 0
def main() -> int:
"""Main entrypoint for the depot_tools ruff wrapper.
Returns:
Exit code of the invoked formatter subprocess.
"""
root_dir, args = extract_root_flag(sys.argv[1:])
if args and args[0] == "--batch":
return run_batch()
# Determine target path for config detection
target_path = os.getcwd()
for i, arg in enumerate(args):
if arg.startswith("--stdin-filename="):
target_path = arg.split("=", 1)[1]
break
elif arg == "--stdin-filename" and i + 1 < len(args):
target_path = args[i + 1]
break
elif not arg.startswith("-") and arg not in KNOWN_SUBCOMMANDS:
target_path = arg
break
try:
parsed_args = parse_ranges(args)
opts = parse_formatting_options(parsed_args.pass_through_args)
if opts.subcommand is not None and opts.subcommand != "format":
return run_ruff_with_ranges(args)
except Exception:
pass
if should_use_ruff(target_path, root_dir):
return run_ruff_with_ranges(args)
else:
# Before attempting YAPF as a backoff, ascertain that a YAPF formatter/config
# is present. If the file was rejected by Ruff (e.g. excluded in ruff.toml)
# and no YAPF config governs the directory, return 0 (no-op).
if not has_yapf_config(target_path, root_dir):
return 0
# Fallback to YAPF
# Find yapf wrapper in the same directory as this script
depot_tools_dir = os.path.dirname(os.path.abspath(__file__))
yapf_script = os.path.join(depot_tools_dir, "yapf")
yapf_args = translate_args(args)
# We must invoke the yapf wrapper via vpython3 so it gets its own env
vpython_bin = "vpython3"
if sys.platform == "win32":
vpython_bin = "vpython3.bat"
cmd = [vpython_bin, yapf_script] + yapf_args
return subprocess.call(cmd)
if __name__ == "__main__":
sys.exit(main())