Coverage for src/pullapprove/config.py: 95%
405 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-10 21:44 -0500
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-10 21:44 -0500
1from __future__ import annotations
3import posixpath
4import re
5import tomllib
6import warnings
8with warnings.catch_warnings():
9 warnings.simplefilter("ignore", DeprecationWarning)
10 import sre_parse
11from collections.abc import Generator
12from enum import StrEnum
13from pathlib import Path
14from typing import Any, Self
16from pydantic import (
17 BaseModel,
18 ConfigDict,
19 Field,
20 RootModel,
21 field_validator,
22 model_validator,
23)
24from wcmatch import glob
26from .checklists import Checklist
28CONFIG_FILENAME_PREFIX = "CODEREVIEW"
30_REPEAT_OPS = {sre_parse.MAX_REPEAT, sre_parse.MIN_REPEAT}
33def _has_nested_quantifiers(data: Any) -> bool:
34 """Detect patterns like (a+)+ that cause catastrophic backtracking."""
35 for op, av in data:
36 if op in _REPEAT_OPS:
37 if _contains_quantifier(av[2]):
38 return True
39 elif op == sre_parse.SUBPATTERN:
40 if _has_nested_quantifiers(av[-1]):
41 return True
42 elif op == sre_parse.BRANCH:
43 if any(_has_nested_quantifiers(branch) for branch in av[1]):
44 return True
45 return False
48def _contains_quantifier(data: Any) -> bool:
49 for op, av in data:
50 if op in _REPEAT_OPS:
51 return True
52 elif op == sre_parse.SUBPATTERN:
53 if _contains_quantifier(av[-1]):
54 return True
55 elif op == sre_parse.BRANCH:
56 if any(_contains_quantifier(branch) for branch in av[1]):
57 return True
58 return False
61CONFIG_FILENAME = "CODEREVIEW.toml"
64_TEAM_REF_SEGMENT = r"[a-zA-Z0-9][a-zA-Z0-9\-_.]*"
65_TEAM_REF_RE = re.compile(rf"^{_TEAM_REF_SEGMENT}(/{_TEAM_REF_SEGMENT})+$")
67# Fields that hold reviewer identities (plain usernames, `$aliases`, and
68# `@team` refs) rather than paths/code/labels. Team refs only expand here.
69USER_LIST_FIELDS = ("authors", "reviewers", "alternates", "cc")
71# Roster fields where "!" performs compile-time subtraction (remove from the
72# resolved list) rather than surviving as a match-time predicate. `authors`
73# is deliberately excluded — its "!" entries are consumed by `matches_author`.
74ROSTER_FIELDS = ("reviewers", "alternates", "cc")
77def _split_team_ref(value: str) -> tuple[str, str] | None:
78 """Split a value into `(prefix, ref)` if it has team-reference shape
79 (`@org/team` or `!@org/team`), `prefix` being `"!"` or `""`. Returns
80 `None` for anything else, so callers fall back to their own handling of
81 plain values.
82 """
83 if value.startswith("!@"):
84 return "!", value[2:]
85 if value.startswith("@"):
86 return "", value[1:]
87 return None
90def is_unexpanded_ref(value: str) -> bool:
91 """True if the value is a `$alias` or `@team` reference (optionally
92 negated with a leading `!`) that has not been expanded — as opposed to a
93 plain username. Only an offline compile (`teams=None`) leaves such values
94 in user-list fields.
95 """
96 return value.removeprefix("!").startswith(("$", "@"))
99def _validate_team_refs(values: list[str]) -> list[str]:
100 """Team references (`@org/team`, `!@org/team`) need at least two
101 slash-separated segments. This only checks shape — membership is resolved
102 later, at compile time, against the caller-provided `teams` mapping.
104 Any other value containing "@" is rejected too — that shape is reserved
105 for team references (and, in future, email-style identifiers), so it
106 can't be confused with a plain username.
107 """
108 for value in values:
109 split = _split_team_ref(value)
110 if split is None:
111 if "@" in value:
112 raise ValueError(
113 f"Invalid value '{value}': email addresses are not "
114 "supported here — use the platform username"
115 )
116 continue
118 _prefix, ref = split
119 if not _TEAM_REF_RE.match(ref):
120 raise ValueError(
121 f"Invalid team reference '{value}': team references need the "
122 "'org/team' form"
123 )
124 return values
127def _expand_team_ref(
128 ref: str, teams: dict[str, list[str]] | None, prefix: str
129) -> list[str]:
130 """Expand a single team reference (without its `@`/`!@`) to member usernames.
132 `teams=None` is the offline/CLI mode: the library never calls out to
133 GitHub/GitLab itself, so with no mapping provided the reference passes
134 through unexpanded rather than erroring. With a `teams` mapping (even an
135 empty one), an unresolvable ref is a loud config error, matching how
136 unknown `$aliases` are handled.
137 """
138 if teams is None:
139 return [f"{prefix}@{ref}"]
141 members = teams.get(ref.lower())
142 if members is None:
143 raise ValueError(f"Unknown team: {prefix}@{ref}")
145 return [f"{prefix}{member}" for member in members]
148def _expand_aliases(
149 values: list[str],
150 aliases: dict[str, list[str]],
151 teams: dict[str, list[str]] | None = None,
152 expand_teams: bool = False,
153 _seen: set[str] | None = None,
154 _path: list[str] | None = None,
155) -> list[str]:
156 """Replace alias references in a list with their mapped values recursively.
158 Team references (`@org/team`) are only expanded when `expand_teams` is
159 True — user-list fields (reviewers, alternates, authors, cc). Elsewhere
160 (paths, code, labels) a leading `@` is a literal string, e.g. npm-style
161 scoped path patterns like `@vendor/pkg/**`. Teams are always leaf nodes —
162 their values are plain usernames, never other refs — so expanding one
163 never recurses further.
164 """
165 if _seen is None:
166 _seen = set()
167 if _path is None:
168 _path = []
170 expanded: list[str] = []
171 for value in values:
172 # Support negated aliases like "!$team" -> ["!alice", "!bob"]
173 if value.startswith("!$"):
174 prefix = "!"
175 alias_ref = value[2:]
176 elif value.startswith("$"):
177 prefix = ""
178 alias_ref = value[1:]
179 elif expand_teams and (team_ref := _split_team_ref(value)) is not None:
180 team_prefix, ref = team_ref
181 expanded.extend(_expand_team_ref(ref=ref, teams=teams, prefix=team_prefix))
182 continue
183 else:
184 expanded.append(value)
185 continue
187 if alias_ref in _seen:
188 # Cycle detected, raise an error with the cycle path
189 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref]
190 raise ValueError(
191 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}"
192 )
193 if alias_ref in aliases:
194 _seen.add(alias_ref)
195 _path.append(alias_ref)
196 # Recursively expand the alias values
197 nested_expanded = _expand_aliases(
198 aliases[alias_ref],
199 aliases=aliases,
200 teams=teams,
201 expand_teams=expand_teams,
202 _seen=_seen,
203 _path=_path,
204 )
205 if prefix:
206 expanded.extend(prefix + v for v in nested_expanded)
207 else:
208 expanded.extend(nested_expanded)
209 _path.pop()
210 _seen.remove(alias_ref)
211 else:
212 # Unknown alias — surface it loudly instead of silently dropping the
213 # reference (a typo'd alias would otherwise vanish reviewers/paths).
214 raise ValueError(f"Unknown alias: {prefix}${alias_ref}")
216 # Remove duplicates while preserving order
217 return list(dict.fromkeys(expanded))
220def _apply_negations(values: list[str]) -> list[str]:
221 """Compile-time subtraction for roster fields (`ROSTER_FIELDS`): a "!name"
222 entry removes "name" from the resolved list instead of surviving as a
223 match-time rule (contrast `authors`, handled by `matches_author`).
225 A list still holding an unexpanded `@team` ref (offline compile, i.e.
226 teams=None) is returned untouched — it isn't fully resolved yet, so
227 subtraction can't run, and the partially-resolved config must round-trip
228 unchanged. `$aliases` must already be expanded by the caller.
230 The wildcard+negation error fires BEFORE that early return: it's a check
231 on the written form (`"*"` alongside any `"!"` entry, team ref or not),
232 so an offline compile must reject it the same way the server will — a
233 `pullapprove check` that passes locally can't then error in production.
234 """
235 if "*" in values and any(v.startswith("!") for v in values):
236 raise ValueError(
237 'Negation cannot be combined with "*" in reviewers/alternates/cc '
238 "(wildcard exclusion is not supported)"
239 )
241 if any(_split_team_ref(v) is not None for v in values):
242 return values
244 negations = {v[1:].lower() for v in values if v.startswith("!")}
245 return [v for v in values if not v.startswith("!") and v.lower() not in negations]
248def _resolve_extends_path(extending_path: str, extends_ref: str) -> str:
249 """Resolve an `extends` reference to a canonical repo-relative config key.
251 - `/x` is repo-root-relative.
252 - everything else (`../x`, `dir/x`, bare `x`) is relative to the extending
253 file's directory.
255 Raises if the reference escapes above the repo root.
256 """
257 if extends_ref.startswith("/"):
258 resolved = posixpath.normpath(extends_ref.lstrip("/"))
259 else:
260 base_dir = posixpath.dirname(extending_path)
261 resolved = posixpath.normpath(posixpath.join(base_dir, extends_ref))
263 if resolved == ".." or resolved.startswith("../"):
264 raise ValueError(
265 f"Invalid extends path: '{extends_ref}' points above the repo root"
266 )
268 return resolved
271def _anchor_path(base_dir: str, pattern: str) -> str:
272 """Anchor a scope path glob at `base_dir` (the owning config's directory).
274 Scope paths are written relative to the config they live in. A leading `/`
275 makes a pattern repo-root-absolute (escape hatch); a leading `!` negation is
276 preserved. With an empty `base_dir` (root config) the pattern is unchanged.
277 """
278 negate = pattern.startswith("!")
279 if negate:
280 pattern = pattern[1:]
282 if pattern.startswith("/"):
283 anchored = pattern.lstrip("/")
284 elif base_dir:
285 anchored = f"{base_dir}/{pattern}"
286 else:
287 anchored = pattern
289 return f"!{anchored}" if negate else anchored
292class OwnershipChoices(StrEnum):
293 EMPTY = ""
294 APPEND = "append"
295 GLOBAL = "global"
298class ScopeModel(BaseModel):
299 model_config = ConfigDict(extra="forbid")
301 # Required fields
302 name: str = Field(min_length=1)
303 paths: list[str] = Field(min_length=1)
305 # Optional fields
307 # Expanded version of lines could be dict
308 # with fnmatch, regex, exclude patterns, etc?
309 code: list[str] = []
311 # This only filtering field that can't be used with raw diff/files...
312 # If we get into that, the others are:
313 # - labels
314 # - ref
315 # - statuses
316 # - dates
317 # - body
318 # - title
319 # - other scopes
320 # (this is how I ended up with expressions...
321 # I'm not trying to build a general purpose workflow tool,
322 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy)
323 authors: list[str] = []
325 # (defaults should be the "empty" values)
326 description: str = ""
327 reviewers: list[str] = []
328 alternates: list[str] = []
329 cc: list[str] = []
331 # Review scoring
332 require: int = 0
333 author_value: int = 0
335 # How scopes are combined
336 ownership: OwnershipChoices = OwnershipChoices.EMPTY
338 # Actionable items
339 request: int = 0
340 labels: list[str] = []
341 instructions: str = ""
343 # Approval checklist
344 checklist: Checklist | None = None
346 @field_validator("name", mode="after")
347 @classmethod
348 def validate_name(cls, name: str) -> str:
349 if "," in name:
350 raise ValueError("Scope name cannot contain commas")
351 return name
353 @field_validator(*USER_LIST_FIELDS, mode="after")
354 @classmethod
355 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
356 return _validate_team_refs(values)
358 @field_validator("code", mode="after")
359 @classmethod
360 def validate_code_patterns(cls, code: list[str]) -> list[str]:
361 for pattern in code:
362 try:
363 parsed = sre_parse.parse(pattern)
364 except re.error as e:
365 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None
366 if _has_nested_quantifiers(parsed):
367 raise ValueError(
368 f"Regex pattern '{pattern}' contains nested quantifiers, "
369 "which can cause catastrophic backtracking."
370 )
371 return code
373 @model_validator(mode="after")
374 def validate_reviewers_for_require(self) -> ScopeModel:
375 all_reviewers = self.reviewers + self.alternates
377 # Skip if wildcard - anyone can review
378 if "*" in all_reviewers:
379 return self
381 # Skip if aliases or team refs (possibly negated with "!") are not yet
382 # expanded. Re-validation after compilation only re-checks refs that
383 # actually resolve — an offline compile (teams=None) leaves refs
384 # unexpanded, so this same skip fires again there too.
385 if any(is_unexpanded_ref(r) for r in all_reviewers):
386 return self
388 if len(all_reviewers) < self.require:
389 raise ValueError(
390 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified"
391 )
392 return self
394 def printed_name(self) -> str:
395 match self.ownership:
396 case OwnershipChoices.APPEND:
397 return "+" + self.name
398 case OwnershipChoices.GLOBAL:
399 return "*" + self.name
401 return self.name
403 def __eq__(self, other: Any) -> bool:
404 return self.name == other.name
406 def matches_path(self, path: Path) -> bool:
407 # TODO paths shouldn't start with /
408 return glob.globmatch(
409 path,
410 self.paths,
411 flags=glob.GLOBSTAR
412 | glob.BRACE
413 | glob.NEGATE
414 | glob.IGNORECASE
415 | glob.DOTGLOB,
416 )
418 def matches_code(self, code: str) -> Generator[dict[str, int]]:
419 patterns = getattr(self, "_code_regex_patterns", [])
420 if not patterns:
421 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
422 self._code_regex_patterns = patterns
424 for pattern in patterns:
425 for match in pattern.finditer(code):
426 start_index = match.start()
427 end_index = match.end()
429 start_line = code.count("\n", 0, start_index) + 1
430 start_col = start_index - code.rfind("\n", 0, start_index)
432 end_line = code.count("\n", 0, end_index) + 1
433 end_col = end_index - code.rfind("\n", 0, end_index)
435 yield {
436 "start_line": start_line,
437 "start_col": start_col,
438 "end_line": end_line,
439 "end_col": end_col,
440 }
442 def matches_author(self, author_username: str) -> bool:
443 if not self.authors:
444 # No authors specified, so assume it matches
445 return True
447 author_username_lower = author_username.lower()
449 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
450 authors = [a.lower() for a in self.authors if not a.startswith("!")]
452 if author_username_lower in negated_authors:
453 # If the author is in the negated list, return False
454 return False
456 if not authors:
457 # Negation-only: everyone not negated matches
458 return True
460 if author_username_lower in authors:
461 # If the author is in the authors list, return True
462 return True
464 return False
467class LargeScaleChangeModel(BaseModel):
468 model_config = ConfigDict(extra="forbid")
470 # Note, an LSC only applies to diffs, not raw files,
471 # because we have to know what *changed*.
473 # Pretty similar to a scope, but more manual.
474 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
475 require: int = 1
476 reviewers: list[str] = [] # Field(min_length=1)
477 # min_paths: int = 300
478 # min_lines: int = 3000
479 labels: list[str] = []
480 # really need author value too...?
482 @field_validator("reviewers", mode="after")
483 @classmethod
484 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
485 return _validate_team_refs(values)
488class ConfigModel(BaseModel):
489 model_config = ConfigDict(extra="forbid")
491 # Nothing is technically required
492 extends: list[str] = []
493 template: bool = False
494 aliases: dict[str, list[str]] = {}
495 large_scale_change: LargeScaleChangeModel | None = None
496 scopes: list[ScopeModel] = []
498 @field_validator("scopes", mode="after")
499 @classmethod
500 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
501 seen: set[str] = set()
502 for scope in scopes:
503 if scope.name.lower() in seen:
504 raise ValueError(f"Duplicate scope name: {scope.name}")
505 seen.add(scope.name.lower())
507 return scopes
509 @field_validator("extends", mode="before")
510 @classmethod
511 def validate_extends(cls, extends: list[str]) -> list[str]:
512 for i, path in enumerate(extends):
513 basename = Path(path).name
514 if not basename.startswith(CONFIG_FILENAME_PREFIX):
515 raise ValueError(
516 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
517 )
518 return extends
520 def compiled_config(
521 self,
522 config_path: Path,
523 other_configs: ConfigModels,
524 teams: dict[str, list[str]] | None = None,
525 ) -> ConfigModel:
526 """
527 Resolve `extends` and replace aliases, returning the effective config.
529 Two phases: flatten the whole extends chain into one merged (raw,
530 unexpanded) config, then expand aliases once. Expanding after the full
531 merge is what makes transitive inheritance and cross-chain alias
532 scoping work — an alias defined anywhere in the chain resolves anywhere.
534 `teams` maps team refs (any case, without the leading `@`) to member
535 usernames, and is only consulted for user-list fields (reviewers,
536 alternates, authors, cc, large_scale_change.reviewers). Keys are
537 lowercased here, so callers don't need to normalize case themselves.
538 With `teams=None` (the default), `@org/team` references in those
539 fields pass through unexpanded — the offline/CLI mode, since this
540 library never calls out to GitHub/GitLab itself. That
541 partially-resolved config is only sanctioned for offline use;
542 anything that needs real reviewer usernames must pass a `teams`
543 mapping.
545 Pure function of the raw `self` and `other_configs` (it never reads its
546 own anchored output), so it is safe to call uncached.
547 """
549 if teams is not None:
550 teams = {key.lower(): members for key, members in teams.items()}
552 compiled_data = self._merged_data(config_path, other_configs)
554 # Expand aliases for any aliasable list fields. Team refs only expand
555 # in user-list fields — paths/code/labels keep a leading "@" literal.
556 for scope in compiled_data["scopes"]:
557 for field in [
558 "paths",
559 "code",
560 "authors",
561 "reviewers",
562 "alternates",
563 "cc",
564 "labels",
565 ]:
566 if field in scope:
567 scope[field] = _expand_aliases(
568 scope[field],
569 compiled_data["aliases"],
570 teams=teams,
571 expand_teams=field in USER_LIST_FIELDS,
572 )
574 # Apply compile-time "!" subtraction to roster fields.
575 # `_apply_negations` leaves a field with an unexpanded team ref
576 # (offline compile, i.e. teams=None) untouched.
577 for scope in compiled_data["scopes"]:
578 for field in ROSTER_FIELDS:
579 if field in scope:
580 scope[field] = _apply_negations(scope[field])
582 # The "*" wildcard is only meaningful in `reviewers` — everywhere
583 # else it's matched as a literal username and silently does
584 # nothing: in `alternates` the scope stays pending forever, in
585 # `authors` the scope never applies to any PR, in `cc` nobody is
586 # notified. Reject it at compile time (after alias expansion, so
587 # `$alias` indirection can't smuggle it in) rather than as a
588 # ScopeModel validator, because compiled configs stored inside
589 # old processing results must keep parsing.
590 # Messages must stay under ~130 chars: they become the git-host
591 # commit status description, which the adapters slice to 140.
592 for field, hint in (
593 (
594 "authors",
595 "remove it (a scope with no authors applies to any author)",
596 ),
597 (
598 "alternates",
599 "add it to reviewers instead (wildcard reviewers are never auto-requested)",
600 ),
601 ("cc", "remove it"),
602 ):
603 if "*" in scope.get(field, []):
604 raise ValueError(
605 f"Scope '{scope['name']}': \"*\" is not supported in "
606 f"{field} — {hint}"
607 )
609 if large_scale_change := compiled_data.get("large_scale_change"):
610 large_scale_change["reviewers"] = _expand_aliases(
611 large_scale_change["reviewers"],
612 compiled_data["aliases"],
613 teams=teams,
614 expand_teams=True,
615 )
616 large_scale_change["labels"] = _expand_aliases(
617 large_scale_change["labels"],
618 compiled_data["aliases"],
619 )
620 large_scale_change["reviewers"] = _apply_negations(
621 large_scale_change["reviewers"]
622 )
624 # Anchor each scope's paths at the directory tagged during flattening
625 # (after alias expansion, so any `$path-alias` is resolved first). The
626 # transient tag is popped so it never reaches the model.
627 for scope in compiled_data["scopes"]:
628 anchor_dir = scope.pop("_anchor_dir", "")
629 scope["paths"] = [_anchor_path(anchor_dir, p) for p in scope["paths"]]
631 # The compiled config is the self-contained effective config: extends
632 # are already merged in and aliases already expanded, so drop both. This
633 # keeps stored results lean and makes the compiled form standalone (it
634 # can never dangle on a missing extends target or re-expand differently).
635 compiled_data["extends"] = []
636 compiled_data["aliases"] = {}
638 return ConfigModel.from_data(
639 data=compiled_data,
640 path=config_path,
641 )
643 def _merged_data(
644 self,
645 config_path: Path,
646 other_configs: ConfigModels,
647 _in_progress: list[str] | None = None,
648 _seen: set[str] | None = None,
649 ) -> dict[str, Any]:
650 """
651 Flatten the `extends` chain into one merged, *unexpanded* config dict.
653 Parents are merged before this config (so a child can specialize), with
654 aliases unioned child-wins and the large-scale-change config taken from
655 the child if set else the first parent that defines one.
657 `_in_progress` is the current ancestor path, used to detect circular
658 extends. `_seen` is every config already merged into this flatten, used
659 to merge a shared ancestor only once (diamond dedup).
660 """
661 if _in_progress is None:
662 _in_progress = []
663 if _seen is None:
664 _seen = set()
666 config_path_str = str(config_path)
667 if config_path_str in _in_progress:
668 cycle = _in_progress[_in_progress.index(config_path_str) :] + [
669 config_path_str
670 ]
671 raise ValueError(
672 f"Circular reference detected in extends: {' -> '.join(cycle)}"
673 )
674 _in_progress.append(config_path_str)
676 inherited_scopes: list[dict[str, Any]] = []
677 inherited_aliases: dict[str, list[str]] = {}
678 inherited_lsc: dict[str, Any] | None = None
680 for extend_path in self.extends:
681 resolved_path = _resolve_extends_path(config_path_str, extend_path)
682 if resolved_path not in other_configs:
683 raise ValueError(
684 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')"
685 )
686 if resolved_path in _seen:
687 # Already merged via another branch (diamond) — skip the dup.
688 continue
690 parent_data = other_configs[resolved_path]._merged_data(
691 Path(resolved_path), other_configs, _in_progress, _seen
692 )
693 inherited_scopes = inherited_scopes + parent_data["scopes"]
694 inherited_aliases = inherited_aliases | parent_data["aliases"]
695 inherited_lsc = inherited_lsc or parent_data["large_scale_change"]
697 merged = self.model_dump()
698 merged["scopes"] = inherited_scopes + merged["scopes"]
699 merged["aliases"] = inherited_aliases | merged["aliases"]
700 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc
702 # Tag each scope with the directory its paths should anchor at. A scope's
703 # paths are relative to the config that owns it, so the first
704 # non-template config to consume a scope claims it: a non-template's own
705 # scopes (and any it inherits from a template) anchor at its directory,
706 # while a template defers to its consumer. `setdefault` means an
707 # already-tagged scope (from a non-template ancestor) keeps its anchor.
708 if not self.template:
709 base_dir = posixpath.dirname(config_path_str)
710 for scope in merged["scopes"]:
711 scope.setdefault("_anchor_dir", base_dir)
713 _seen.add(config_path_str)
714 _in_progress.pop()
716 return merged
718 @classmethod
719 def from_filesystem(cls, path: Path | str) -> ConfigModel:
720 with open(path, "rb") as f:
721 return cls.from_data(tomllib.load(f), path)
723 @classmethod
724 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
725 return cls.from_data(tomllib.loads(content), path)
727 @classmethod
728 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
729 return cls(**data)
732class _ConfigModelsBase(RootModel):
733 """Shared storage and accessors for a set of configs keyed by repo path."""
735 root: dict[str, ConfigModel]
737 @classmethod
738 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self:
739 """Build from a dict of already-constructed configs keyed by path."""
740 configs = cls(root={})
741 for path, config_model in models.items():
742 configs.root[str(Path(path))] = config_model
743 return configs
745 def get_config_models(self) -> dict[str, ConfigModel]:
746 return dict(self.root.items())
748 def __bool__(self) -> bool:
749 return bool(self.root)
751 def __getitem__(self, key: str) -> ConfigModel:
752 return self.root[key]
754 def __contains__(self, key: str) -> bool:
755 return key in self.root
757 def __len__(self) -> int:
758 return len(self.root)
761class ConfigModels(_ConfigModelsBase):
762 """Configs exactly as loaded from the repo — extends unresolved, aliases
763 unexpanded, paths unanchored. Build the set up, then call `compiled()`."""
765 @classmethod
766 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
767 """Load configs from a dict of parsed config data keyed by path."""
768 configs = cls(root={})
770 for path, config_data in data.items():
771 config = ConfigModel.from_data(config_data, Path(path))
772 configs.add_config(config, Path(path))
774 return configs
776 def add_config(self, config: ConfigModel, path: Path) -> None:
777 self.root[str(path)] = config
779 def team_refs(self) -> set[str]:
780 """Collect every team ref (lowercase, no leading `@`/`!`) that
781 `compiled(teams=...)` would actually try to expand: refs written
782 directly in a user-list field (scopes' `USER_LIST_FIELDS` and
783 `large_scale_change.reviewers`), plus any refs reachable from those
784 fields through `$alias`/`!$alias` chains.
786 Meant for callers that need to know which teams to fetch/sync before
787 calling `compiled(teams=...)`.
789 Implemented as an offline compile (`teams=None`): aliases expand but
790 team refs pass through unexpanded, so whatever refs remain in the
791 compiled user-list fields are — by construction — exactly the refs a
792 real compile will try to expand. A ref that only appears in a
793 non-user-list field (e.g. an npm-style `@vendor/pkg/**` in `paths`)
794 or inside an alias nothing references never survives into a compiled
795 user-list field, so it is never collected. Raises the same config
796 errors `compiled()` would (unknown alias, circular refs, ...), just
797 earlier.
798 """
799 # Cheap pre-check: a team ref can only enter a compile as a literal
800 # `@`/`!@` value in a user-list field or an alias value. Most repos
801 # have none, and skipping the compile keeps this near-free for them.
802 candidate_lists: list[list[str]] = []
803 for config in self.root.values():
804 candidate_lists.extend(config.aliases.values())
805 for scope in config.scopes:
806 for field in USER_LIST_FIELDS:
807 candidate_lists.append(getattr(scope, field))
808 if config.large_scale_change:
809 candidate_lists.append(config.large_scale_change.reviewers)
810 if not any(
811 _split_team_ref(value) is not None
812 for values in candidate_lists
813 for value in values
814 ):
815 return set()
817 refs: set[str] = set()
819 def collect_refs(values: list[str]) -> None:
820 for value in values:
821 if (split := _split_team_ref(value)) is not None:
822 refs.add(split[1].lower())
824 for config in self.compiled(teams=None).get_config_models().values():
825 if config.template:
826 continue
827 for scope in config.scopes:
828 for field in USER_LIST_FIELDS:
829 collect_refs(getattr(scope, field))
830 if config.large_scale_change:
831 collect_refs(config.large_scale_change.reviewers)
833 return refs
835 def compiled(
836 self, teams: dict[str, list[str]] | None = None
837 ) -> CompiledConfigModels:
838 """Resolve the whole set into its effective, PR-independent form.
840 Each non-template config is compiled once — extends merged, aliases
841 expanded, paths anchored. Templates are NOT compiled standalone: a
842 template scope may reference an alias the consuming config provides, and
843 its paths anchor at the consumer. They are carried through untouched
844 (folded into each consumer during that consumer's compile, and kept in
845 the set for display).
847 `teams` maps team refs (any case, without the leading `@`) to member
848 usernames; passed straight through to each config's `compiled_config`
849 (see there for case normalization and the `teams=None` vs provided
850 semantics).
852 The result is an immutable `CompiledConfigModels` — there is no way to
853 compile it again, so the non-idempotent path anchoring can never
854 double-apply.
855 """
856 effective: dict[str, ConfigModel] = {}
857 for path, config in self.root.items():
858 if config.template:
859 effective[path] = config
860 else:
861 effective[path] = config.compiled_config(
862 config_path=Path(path), other_configs=self, teams=teams
863 )
865 return CompiledConfigModels.from_config_models(effective)
868class CompiledConfigModels(_ConfigModelsBase):
869 """The effective configs used for matching: every non-template config is
870 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled."""
872 def closest_config(self, file_path: Path) -> ConfigModel:
873 """Return the closest non-template config governing this file."""
874 for parent in file_path.parents:
875 parent_config_path = str(parent / CONFIG_FILENAME)
877 if parent_config_path in self.root:
878 config = self.root[parent_config_path]
880 if config.template:
881 # Skip templates
882 continue
884 return config
886 raise ValueError(f"No config found for {file_path}")
888 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
889 """The primary (repo-root) config's large-scale-change section, if any.
891 The primary was compiled by `compiled()`, so its reviewers/labels are
892 already alias-expanded (e.g. ["$backend"] -> usernames). A `template =
893 true` repo root is a misconfiguration (templates are meant to be
894 extended, not be the primary); it is passed through uncompiled, so its
895 LSC would read with aliases unexpanded.
896 """
897 if CONFIG_FILENAME in self.root:
898 if lsc := self.root[CONFIG_FILENAME].large_scale_change:
899 return lsc
901 return LargeScaleChangeModel()
903 def filter_for_pullrequest(self, author_username: str) -> CompiledConfigModels:
904 """
905 Overlay PR-dependent scope gating: drop scopes that author rules disable
906 for this pull request.
908 This is the only PR-dependent step. The configs are already compiled, so
909 each config's scopes are self-contained and dropping one is a plain list
910 filter — no re-inheritance. Templates are passed through (they are never
911 matched directly; their scopes already live in each consumer).
912 """
913 effective: dict[str, ConfigModel] = {}
914 for config_path, config in self.root.items():
915 if config.template:
916 # Templates are never matched directly; pass them through.
917 effective[config_path] = config
918 continue
920 kept_scopes = [
921 scope
922 for scope in config.scopes
923 if scope.matches_author(author_username)
924 ]
925 effective[config_path] = config.model_copy(update={"scopes": kept_scopes})
927 return CompiledConfigModels.from_config_models(effective)