1from collections.abc import Callable, Generator, Iterable
2from typing import Any, TypeVar
3
4from plain.runtime import settings
5
6from .checks import PreflightCheck
7from .results import PreflightResult, unused_silenced_results
8
9T = TypeVar("T")
10
11UNUSED_SILENCES_CHECK_NAME = "preflight.unused_silences"
12
13
14class CheckUnusedSilences(PreflightCheck):
15 """Reports `PREFLIGHT_SILENCED_RESULTS` entries that matched nothing.
16
17 An unused entry is either a typo or stale — the issue it silenced has
18 been fixed. Not registered like a normal check: it needs every other
19 check's results, so the registry runs it last with the full run's
20 results, and only on a full run (deploy checks included) — a partial
21 run skips checks whose entries would then look unused.
22 """
23
24 def __init__(self, run_results: list[PreflightResult]) -> None:
25 self.run_results = run_results
26
27 def run(self) -> list[PreflightResult]:
28 return [
29 PreflightResult(
30 fix=f"Silenced result {entry!r} matched nothing in this run. "
31 "Remove it from PREFLIGHT_SILENCED_RESULTS or fix the typo.",
32 obj=entry,
33 id="preflight.unused_silence",
34 warning=True,
35 )
36 for entry in unused_silenced_results(self.run_results)
37 ]
38
39
40class CheckRegistry:
41 def __init__(self) -> None:
42 self.checks: dict[
43 str, tuple[type[Any], bool]
44 ] = {} # name -> (check_class, deploy)
45
46 def register_check(
47 self, check_class: type[Any], name: str, deploy: bool = False
48 ) -> None:
49 """Register a check class with a unique name."""
50 if name in self.checks:
51 raise ValueError(f"Check {name} already registered")
52 self.checks[name] = (check_class, deploy)
53
54 def run_checks(
55 self,
56 include_deploy_checks: bool = False,
57 ) -> Generator[tuple[type[Any], str, list[PreflightResult]]]:
58 """
59 Run all registered checks and yield (check_class, name, results) tuples.
60 """
61 # Validate silenced check names (the unused-silences check isn't in
62 # self.checks — the registry emits it itself — but it's silenceable
63 # like any other check)
64 silenced_checks = settings.PREFLIGHT_SILENCED_CHECKS
65 known_checks = set(self.checks.keys()) | {UNUSED_SILENCES_CHECK_NAME}
66 unknown_silenced = set(silenced_checks) - known_checks
67 if unknown_silenced:
68 unknown_names = ", ".join(sorted(unknown_silenced))
69 raise ValueError(
70 f"Unknown check names in PREFLIGHT_SILENCED_CHECKS: {unknown_names}. "
71 "Check for typos or remove outdated check names."
72 )
73
74 all_results: list[PreflightResult] = []
75
76 for name, (check_class, deploy) in sorted(self.checks.items()):
77 # Skip silenced checks
78 if name in silenced_checks:
79 continue
80
81 # Skip deployment checks if not requested
82 if deploy and not include_deploy_checks:
83 continue
84
85 # Instantiate and run check
86 check = check_class()
87 results = check.run()
88 all_results.extend(results)
89 yield check_class, name, results
90
91 if include_deploy_checks and UNUSED_SILENCES_CHECK_NAME not in silenced_checks:
92 check = CheckUnusedSilences(all_results)
93 yield CheckUnusedSilences, UNUSED_SILENCES_CHECK_NAME, check.run()
94
95
96checks_registry = CheckRegistry()
97
98
99def register_check(name: str, *, deploy: bool = False) -> Callable[[type[T]], type[T]]:
100 """
101 Decorator to register a check class.
102
103 Usage:
104 @register_check("security.secret_key", deploy=True)
105 class CheckSecretKey(PreflightCheck):
106 pass
107
108 @register_check("files.upload_temp_dir")
109 class CheckUploadTempDir(PreflightCheck):
110 pass
111 """
112
113 def wrapper(cls: type[T]) -> type[T]:
114 checks_registry.register_check(cls, name=name, deploy=deploy)
115 return cls
116
117 return wrapper
118
119
120run_checks = checks_registry.run_checks
121
122# Cached error/warning counts — populated on first call, refreshed by
123# PreflightView when the full page is viewed.
124_check_counts: dict[str, int] | None = None
125
126
127def count_results_by_severity(
128 checks: Iterable[tuple[Any, str, list[PreflightResult]]],
129) -> tuple[int, int]:
130 """Tally ``(errors, warnings)`` across checks by each check's *visible*
131 (non-silenced) issues. A check with no visible issues counts as neither —
132 an error if any visible issue isn't a warning, otherwise a warning."""
133 error_count = 0
134 warning_count = 0
135 for _check_class, _name, results in checks:
136 visible = [r for r in results if not r.is_silenced()]
137 if not visible:
138 continue
139 if any(not r.warning for r in visible):
140 error_count += 1
141 else:
142 warning_count += 1
143 return error_count, warning_count
144
145
146def get_check_counts() -> dict[str, int]:
147 """Return ``{"errors": N, "warnings": N}``, caching for the process lifetime."""
148 global _check_counts
149
150 if _check_counts is not None:
151 return _check_counts
152
153 from plain.packages import packages_registry
154
155 packages_registry.autodiscover_modules("preflight", include_app=True)
156
157 error_count, warning_count = count_results_by_severity(
158 run_checks(include_deploy_checks=not settings.DEBUG)
159 )
160
161 _check_counts = {"errors": error_count, "warnings": warning_count}
162 return _check_counts
163
164
165def set_check_counts(*, errors: int, warnings: int) -> None:
166 """Update the cached counts (called by PreflightView after running full checks)."""
167 global _check_counts
168 _check_counts = {"errors": errors, "warnings": warnings}