Gather every stub in the allowlist plus any re-export targets they pull in transitively (e.g. os.path -> posixpath -> genericpath). Returns (modules, star_imports_per_module).
(stdlib_root: Path)
| 199 | |
| 200 | |
| 201 | def collect_all_stubs(stdlib_root: Path) -> tuple[dict[str, ModuleStubs], dict[str, list[StarReExport]]]: |
| 202 | """Gather every stub in the allowlist plus any re-export targets they |
| 203 | pull in transitively (e.g. os.path -> posixpath -> genericpath). |
| 204 | Returns (modules, star_imports_per_module). """ |
| 205 | modules: dict[str, ModuleStubs] = {} |
| 206 | star_imports: dict[str, list[StarReExport]] = {} |
| 207 | |
| 208 | # First pass: walk allowlist |
| 209 | queue: list[tuple[Path, str]] = [] |
| 210 | for path in sorted(stdlib_root.rglob("*.pyi")): |
| 211 | rel = path.relative_to(stdlib_root) |
| 212 | parts = list(rel.parts) |
| 213 | if parts[-1] == "__init__.pyi": |
| 214 | parts.pop() |
| 215 | else: |
| 216 | parts[-1] = parts[-1][:-len(".pyi")] |
| 217 | if not parts: |
| 218 | continue |
| 219 | top = parts[0] |
| 220 | if top.startswith("_"): |
| 221 | continue |
| 222 | if top not in ALLOWED_MODULES: |
| 223 | continue |
| 224 | module_qn = ".".join(parts) |
| 225 | queue.append((path, module_qn)) |
| 226 | |
| 227 | seen_targets: set[str] = set() |
| 228 | while queue: |
| 229 | path, mod_qn = queue.pop() |
| 230 | if mod_qn in modules: |
| 231 | continue |
| 232 | ms, stars = parse_module(path, mod_qn) |
| 233 | modules[mod_qn] = ms |
| 234 | star_imports[mod_qn] = stars |
| 235 | # Enqueue re-export targets (e.g. posixpath, ntpath, genericpath) |
| 236 | for star in stars: |
| 237 | tgt = star.target_module |
| 238 | if tgt in seen_targets: |
| 239 | continue |
| 240 | seen_targets.add(tgt) |
| 241 | tgt_path = stdlib_root / (tgt.replace(".", "/") + ".pyi") |
| 242 | if tgt_path.is_file(): |
| 243 | queue.append((tgt_path, tgt)) |
| 244 | else: |
| 245 | tgt_init = stdlib_root / tgt.replace(".", "/") / "__init__.pyi" |
| 246 | if tgt_init.is_file(): |
| 247 | queue.append((tgt_init, tgt)) |
| 248 | |
| 249 | return modules, star_imports |
| 250 | |
| 251 | |
| 252 | def resolve_reexports(modules: dict[str, ModuleStubs], |
no test coverage detected