Coverage for sygaldry / checker.py: 66%
107 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-15 18:36 -0400
1from __future__ import annotations
3__author__ = "Rohan B. Dalton"
5import json
6import os
7import re
8import shutil
9import subprocess
10import tempfile
11from dataclasses import dataclass
12from pathlib import Path
13from typing import Any, Optional
15from .codegen import SourceMapping, generate_check_source
16from .loader import load_config
18_MYPY_PATTERN = re.compile(
19 r"^.+?:(?P<line>\d+):\d+:\s*(?P<severity>error|warning|note):\s*(?P<message>.+?)(?:\s+\[.+\])?$"
20)
21_TY_LINE_PATTERN = re.compile(r"^\s*-->\s*.+?:(?P<line>\d+):\d+")
22_TY_DIAG_PATTERN = re.compile(r"^(?P<severity>error|warning)\[.+?\]:\s*(?P<message>.+)")
23SUPPORTED_CHECKERS = ("ty", "basedpyright", "pyright", "mypy")
26@dataclass(frozen=True)
27class CheckError:
28 """
29 A type error found in a config file.
30 """
32 config_path: str
33 message: str
34 severity: str = "error"
37def check(
38 path: str | Path | None = None,
39 *,
40 type_checker: str | None = None,
41 config: dict[str, Any] | None = None,
42) -> list[CheckError]:
43 """
44 Type check a sygaldry config file.
46 Either *path* or *config* must be provided. When *config* is given
47 the config dict is used directly (useful from the CLI after loading
48 and applying overrides). When *path* is given, the file is loaded
49 via :func:`load_config`.
51 :param path: Path to a YAML/TOML config file.
52 :param type_checker: Which checker to run (``ty``, ``basedpyright``,
53 ``pyright``, or ``mypy``). Auto-detected when *None*.
54 :param config: Pre-loaded config dict.
55 :returns: List of type errors found.
56 """
57 if config is None:
58 if path is None:
59 raise ValueError("Either path or config must be provided.")
60 else:
61 config = load_config(Path(path))
63 source, mappings = generate_check_source(config)
65 checker = type_checker or _detect_type_checker()
66 return _run_and_parse(checker, source, mappings)
69def _detect_type_checker() -> str:
70 """
71 Auto-detect an available type checker.
72 """
73 for name in SUPPORTED_CHECKERS: 73 ↛ 76line 73 didn't jump to line 76 because the loop on line 73 didn't complete
74 if shutil.which(name): 74 ↛ 73line 74 didn't jump to line 73 because the condition on line 74 was always true
75 return name
76 raise RuntimeError(
77 "No supported type checker found. Install ty, basedpyright, pyright, or mypy."
78 )
81def _run_and_parse(
82 checker: str,
83 source: str,
84 mappings: list[SourceMapping],
85) -> list[CheckError]:
86 """
87 Write source to a temp file, run the checker, and parse results.
88 """
89 fd, tmp_path = tempfile.mkstemp(suffix=".py", prefix="_sygaldry_check_")
90 with os.fdopen(fd, "w") as fh:
91 fh.write(source)
92 result = _invoke_checker(checker, tmp_path)
93 os.unlink(tmp_path)
94 return _parse_output(checker, result, mappings)
97def _invoke_checker(checker: str, filepath: str) -> subprocess.CompletedProcess:
98 """
99 Run the type checker subprocess.
100 """
101 if checker in {"pyright", "basedpyright"}: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true
102 cmd = [checker, "--outputjson", filepath]
103 elif checker == "mypy": 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true
104 cmd = ["mypy", "--no-color-output", "--show-column-numbers", filepath]
105 elif checker == "ty": 105 ↛ 108line 105 didn't jump to line 108 because the condition on line 105 was always true
106 cmd = ["ty", "check", filepath]
107 else:
108 raise ValueError(f"Unsupported type checker: {checker}")
110 return subprocess.run(
111 cmd,
112 capture_output=True,
113 text=True,
114 timeout=120,
115 )
118def _parse_output(
119 checker: str,
120 result: subprocess.CompletedProcess,
121 mappings: list[SourceMapping],
122) -> list[CheckError]:
123 """
124 Parse type checker output into CheckError instances.
125 """
126 if checker in ("pyright", "basedpyright"): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 return _parse_pyright(result, mappings)
128 elif checker == "mypy": 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 return _parse_mypy(result, mappings)
130 elif checker == "ty": 130 ↛ 132line 130 didn't jump to line 132 because the condition on line 130 was always true
131 return _parse_ty(result, mappings)
132 return list()
135def _line_to_config_path(line: int, mappings: list[SourceMapping]) -> str:
136 """
137 Map a generated line number to its config path.
138 """
139 best: SourceMapping | None = None
140 for mapping in mappings:
141 if best is None or best.line < mapping.line <= line: 141 ↛ 140line 141 didn't jump to line 140 because the condition on line 141 was always true
142 best = mapping
143 return best.config_path if best else "<unknown>"
146def _parse_pyright(
147 result: subprocess.CompletedProcess,
148 mappings: list[SourceMapping],
149) -> list[CheckError]:
150 errors: list[CheckError] = list()
151 try:
152 data = json.loads(result.stdout)
153 except (json.JSONDecodeError, ValueError):
154 return errors
156 for diag in data.get("generalDiagnostics", list()):
157 line = diag.get("range", {}).get("start", {}).get("line", 0)
158 # pyright uses 0-based lines
159 config_path = _line_to_config_path(line + 1, mappings)
160 severity = diag.get("severity", "error")
161 message = diag.get("message", "")
162 errors.append(
163 CheckError(
164 config_path=config_path,
165 message=message,
166 severity=severity,
167 )
168 )
169 return errors
172def _parse_mypy(
173 result: subprocess.CompletedProcess,
174 mappings: list[SourceMapping],
175) -> list[CheckError]:
176 errors: list[CheckError] = list()
177 for raw_line in result.stdout.splitlines():
178 if match := _MYPY_PATTERN.match(raw_line):
179 line_no = int(match.group("line"))
180 severity = match.group("severity")
181 message = match.group("message")
182 if severity == "note":
183 continue
184 config_path = _line_to_config_path(line_no, mappings)
185 errors.append(
186 CheckError(
187 config_path=config_path,
188 message=message,
189 severity=severity,
190 )
191 )
192 return errors
195def _parse_ty(
196 result: subprocess.CompletedProcess,
197 mappings: list[SourceMapping],
198) -> list[CheckError]:
199 errors: list[CheckError] = list()
200 output = result.stdout + result.stderr
201 lines = output.splitlines()
203 for idx, line in enumerate(lines):
204 if diag_match := _TY_DIAG_PATTERN.match(line):
205 severity = diag_match.group("severity")
206 message = diag_match.group("message")
207 line_no = _extract_ty_line(lines, idx)
208 config_path = _line_to_config_path(line_no, mappings) if line_no else "<unknown>"
209 errors.append(
210 CheckError(
211 config_path=config_path,
212 message=message,
213 severity=severity,
214 )
215 )
217 return errors
220def _extract_ty_line(lines: list[str], diag_idx: int) -> int | None:
221 """
222 Look ahead from a ty diagnostic line to find the --> line number.
223 """
224 for idx in range(diag_idx + 1, min(diag_idx + 6, len(lines))): 224 ↛ 227line 224 didn't jump to line 227 because the loop on line 224 didn't complete
225 if match := _TY_LINE_PATTERN.match(lines[idx]): 225 ↛ 224line 225 didn't jump to line 224 because the condition on line 225 was always true
226 return int(match.group("line"))
227 return None