1from __future__ import annotations
2
3import json
4import subprocess
5import sys
6import tomllib
7from pathlib import Path
8from typing import Any
9
10import click
11from plain.cli import register_cli
12from plain.cli.print import print_event
13from plain.cli.runtime import common_command, without_runtime_setup
14
15from .annotations import AnnotationResult, check_annotations
16from .oxc import OxcTool, install_oxc
17
18DEFAULT_RUFF_CONFIG = Path(__file__).parent / "ruff_defaults.toml"
19
20
21@without_runtime_setup
22@register_cli("code")
23@click.group()
24def cli() -> None:
25 """Code formatting and linting"""
26
27
28@without_runtime_setup
29@cli.command()
30@click.option("--force", is_flag=True, help="Reinstall even if up to date")
31@click.pass_context
32def install(ctx: click.Context, force: bool) -> None:
33 """Install or update oxlint and oxfmt binaries"""
34 config = get_code_config()
35
36 if not config.get("oxc", {}).get("enabled", True):
37 click.secho("Oxc is disabled in configuration", fg="yellow")
38 return
39
40 oxlint = OxcTool("oxlint")
41 oxfmt = OxcTool("oxfmt")
42
43 if force or not (oxlint.is_installed() and oxfmt.is_installed()):
44 version_to_install = config.get("oxc", {}).get("version", "")
45 if version_to_install:
46 click.secho(
47 f"Installing oxlint and oxfmt {version_to_install}...",
48 bold=True,
49 nl=False,
50 )
51 installed = install_oxc(version_to_install)
52 click.secho(f"oxlint and oxfmt {installed} installed", fg="green")
53 else:
54 ctx.invoke(update)
55 else:
56 click.secho("oxlint and oxfmt already installed", fg="green")
57
58
59@without_runtime_setup
60@cli.command()
61def update() -> None:
62 """Update oxlint and oxfmt to latest version"""
63 config = get_code_config()
64
65 if not config.get("oxc", {}).get("enabled", True):
66 click.secho("Oxc is disabled in configuration", fg="yellow")
67 return
68
69 click.secho("Updating oxlint and oxfmt...", bold=True)
70 version = install_oxc()
71 click.secho(f"oxlint and oxfmt {version} installed", fg="green")
72
73
74def _partition_paths(paths: tuple[str, ...]) -> tuple[tuple[str, ...], tuple[str, ...]]:
75 """Split paths into (python_paths, other_paths).
76
77 Directories go into both groups. Files are routed by extension.
78 """
79 python_paths: list[str] = []
80 other_paths: list[str] = []
81 for p in paths:
82 if Path(p).is_dir():
83 python_paths.append(p)
84 other_paths.append(p)
85 elif Path(p).suffix == ".py":
86 python_paths.append(p)
87 else:
88 other_paths.append(p)
89 return tuple(python_paths), tuple(other_paths)
90
91
92@without_runtime_setup
93@cli.command()
94@click.pass_context
95@click.argument("paths", nargs=-1)
96@click.option("--skip-ruff", is_flag=True, help="Skip Ruff checks")
97@click.option("--skip-ty", is_flag=True, help="Skip ty type checks")
98@click.option("--skip-oxc", is_flag=True, help="Skip oxlint and oxfmt checks")
99@click.option("--skip-annotations", is_flag=True, help="Skip type annotation checks")
100def check(
101 ctx: click.Context,
102 paths: tuple[str, ...],
103 skip_ruff: bool,
104 skip_ty: bool,
105 skip_oxc: bool,
106 skip_annotations: bool,
107) -> None:
108 """Check for formatting and linting issues"""
109 if not paths:
110 paths = (".",)
111
112 python_paths, other_paths = _partition_paths(paths)
113 ruff_args = ["--config", str(DEFAULT_RUFF_CONFIG)]
114 config = get_code_config()
115
116 for e in config.get("exclude", []):
117 ruff_args.extend(["--exclude", e])
118
119 def maybe_exit(return_code: int) -> None:
120 if return_code != 0:
121 click.secho(
122 "\nCode check failed. Run `plain fix` and/or fix issues manually.",
123 fg="red",
124 err=True,
125 )
126 sys.exit(return_code)
127
128 if not skip_ruff and python_paths:
129 print_event("ruff check...", newline=False)
130 result = subprocess.run(
131 ["ruff", "check", *python_paths, *ruff_args], check=False
132 )
133 maybe_exit(result.returncode)
134
135 print_event("ruff format --check...", newline=False)
136 result = subprocess.run(
137 ["ruff", "format", *python_paths, "--check", *ruff_args], check=False
138 )
139 maybe_exit(result.returncode)
140
141 if not skip_ty and python_paths and config.get("ty", {}).get("enabled", True):
142 print_event("ty check...", newline=False)
143 ty_args = ["ty", "check", *python_paths, "--no-progress"]
144 for e in config.get("exclude", []):
145 ty_args.extend(["--exclude", e])
146 result = subprocess.run(ty_args, check=False)
147 maybe_exit(result.returncode)
148
149 if not skip_oxc and other_paths and config.get("oxc", {}).get("enabled", True):
150 oxlint = OxcTool("oxlint")
151 oxfmt = OxcTool("oxfmt")
152
153 if not (oxlint.is_installed() and oxfmt.is_installed()):
154 ctx.invoke(install)
155
156 print_event("oxlint...", newline=False)
157 result = oxlint.invoke(*other_paths)
158 maybe_exit(result.returncode)
159
160 print_event("oxfmt --check...", newline=False)
161 result = oxfmt.invoke("--check", *other_paths)
162 maybe_exit(result.returncode)
163
164 if (
165 not skip_annotations
166 and python_paths
167 and config.get("annotations", {}).get("enabled", True)
168 ):
169 print_event("annotations...", newline=False)
170 # Combine top-level exclude with annotation-specific exclude
171 exclude_patterns = list(config.get("exclude", []))
172 exclude_patterns.extend(config.get("annotations", {}).get("exclude", []))
173 ann_result = check_annotations(
174 *python_paths, exclude_patterns=exclude_patterns or None
175 )
176 if ann_result.missing_count > 0:
177 click.secho(
178 f"{ann_result.missing_count} functions are untyped",
179 fg="red",
180 )
181 click.secho("Run 'plain code annotations --details' for details")
182 maybe_exit(1)
183 else:
184 click.secho("All functions typed!", fg="green")
185
186
187@without_runtime_setup
188@cli.command()
189@click.argument("paths", nargs=-1)
190@click.option("--details", is_flag=True, help="List untyped functions")
191@click.option("--json", "as_json", is_flag=True, help="Output as JSON")
192def annotations(paths: tuple[str, ...], details: bool, as_json: bool) -> None:
193 """Check type annotation status"""
194 if not paths:
195 paths = (".",)
196 config = get_code_config()
197 # Combine top-level exclude with annotation-specific exclude
198 exclude_patterns = list(config.get("exclude", []))
199 exclude_patterns.extend(config.get("annotations", {}).get("exclude", []))
200 result = check_annotations(*paths, exclude_patterns=exclude_patterns or None)
201 if as_json:
202 _print_annotations_json(result)
203 else:
204 _print_annotations_report(result, show_details=details)
205
206
207def _print_annotations_report(
208 result: AnnotationResult,
209 show_details: bool = False,
210) -> None:
211 """Print the annotation report with colors."""
212 if result.total_functions == 0:
213 click.echo("No functions found")
214 return
215
216 # Detailed output first (if enabled and there are untyped functions)
217 if show_details and result.missing_count > 0:
218 # Collect all untyped functions with full paths
219 untyped_items: list[tuple[str, str, int, list[str]]] = []
220
221 for stats in result.file_stats:
222 for func in stats.functions:
223 if not func.is_fully_typed:
224 issues = []
225 if not func.has_return_type:
226 issues.append("return type")
227 missing_params = func.total_params - func.typed_params
228 if missing_params > 0:
229 param_word = "param" if missing_params == 1 else "params"
230 issues.append(f"{missing_params} {param_word}")
231 untyped_items.append((stats.path, func.name, func.line, issues))
232
233 # Sort by file path, then line number
234 untyped_items.sort(key=lambda x: (x[0], x[2]))
235
236 # Print each untyped function
237 for file_path, func_name, line, issues in untyped_items:
238 location = click.style(f"{file_path}:{line}", fg="cyan")
239 issue_str = click.style(f"({', '.join(issues)})", dim=True)
240 click.echo(f"{location} {func_name} {issue_str}")
241
242 click.echo()
243
244 # Summary line
245 pct = result.coverage_percentage
246 color = "green" if result.missing_count == 0 else "red"
247 click.secho(
248 f"{pct:.1f}% typed ({result.fully_typed_functions}/{result.total_functions} functions)",
249 fg=color,
250 )
251
252 # Code smell indicators (only if present)
253 smells = []
254 if result.total_ignores > 0:
255 smells.append(f"{result.total_ignores} ignore")
256 if result.total_casts > 0:
257 smells.append(f"{result.total_casts} cast")
258 if result.total_asserts > 0:
259 smells.append(f"{result.total_asserts} assert")
260 if smells:
261 click.secho(f"{', '.join(smells)}", fg="yellow")
262
263
264def _print_annotations_json(result: AnnotationResult) -> None:
265 """Print the annotation report as JSON."""
266 output = {
267 "overall_coverage": result.coverage_percentage,
268 "total_functions": result.total_functions,
269 "fully_typed_functions": result.fully_typed_functions,
270 "total_ignores": result.total_ignores,
271 "total_casts": result.total_casts,
272 "total_asserts": result.total_asserts,
273 }
274 click.echo(json.dumps(output))
275
276
277@common_command
278@without_runtime_setup
279@register_cli("fix", shortcut_for="code fix")
280@cli.command()
281@click.pass_context
282@click.argument("paths", nargs=-1)
283@click.option("--unsafe-fixes", is_flag=True, help="Apply ruff unsafe fixes")
284@click.option("--add-noqa", is_flag=True, help="Add noqa comments to suppress errors")
285def fix(
286 ctx: click.Context, paths: tuple[str, ...], unsafe_fixes: bool, add_noqa: bool
287) -> None:
288 """Fix formatting and linting issues"""
289 if not paths:
290 paths = (".",)
291
292 python_paths, other_paths = _partition_paths(paths)
293 ruff_args = ["--config", str(DEFAULT_RUFF_CONFIG)]
294 config = get_code_config()
295
296 for e in config.get("exclude", []):
297 ruff_args.extend(["--exclude", e])
298
299 if unsafe_fixes and add_noqa:
300 raise click.UsageError("Cannot use both --unsafe-fixes and --add-noqa")
301
302 if python_paths:
303 if unsafe_fixes:
304 print_event("ruff check --fix --unsafe-fixes...", newline=False)
305 result = subprocess.run(
306 ["ruff", "check", *python_paths, "--fix", "--unsafe-fixes", *ruff_args],
307 check=False,
308 )
309 elif add_noqa:
310 print_event("ruff check --add-noqa...", newline=False)
311 result = subprocess.run(
312 ["ruff", "check", *python_paths, "--add-noqa", *ruff_args], check=False
313 )
314 else:
315 print_event("ruff check --fix...", newline=False)
316 result = subprocess.run(
317 ["ruff", "check", *python_paths, "--fix", *ruff_args], check=False
318 )
319
320 if result.returncode != 0:
321 sys.exit(result.returncode)
322
323 print_event("ruff format...", newline=False)
324 result = subprocess.run(
325 ["ruff", "format", *python_paths, *ruff_args], check=False
326 )
327 if result.returncode != 0:
328 sys.exit(result.returncode)
329
330 if other_paths and config.get("oxc", {}).get("enabled", True):
331 oxlint = OxcTool("oxlint")
332 oxfmt = OxcTool("oxfmt")
333
334 if not (oxlint.is_installed() and oxfmt.is_installed()):
335 ctx.invoke(install)
336
337 if unsafe_fixes:
338 print_event("oxlint --fix-dangerously...", newline=False)
339 result = oxlint.invoke(*other_paths, "--fix-dangerously")
340 else:
341 print_event("oxlint --fix...", newline=False)
342 result = oxlint.invoke(*other_paths, "--fix")
343
344 if result.returncode != 0:
345 sys.exit(result.returncode)
346
347 print_event("oxfmt...", newline=False)
348 result = oxfmt.invoke(*other_paths)
349
350 if result.returncode != 0:
351 sys.exit(result.returncode)
352
353
354def get_code_config() -> dict[str, Any]:
355 pyproject = Path("pyproject.toml")
356 if not pyproject.exists():
357 return {}
358 with pyproject.open("rb") as f:
359 return tomllib.load(f).get("tool", {}).get("plain", {}).get("code", {})