Coverage for src/lpip/catalog.py: 88%
203 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-17 20:19 +0000
1"""Load and look up packages in lpip-sources.toml."""
3from __future__ import annotations
5import re
6import tomllib
7from dataclasses import dataclass, replace
8from pathlib import Path
10import tomlkit
11from packaging.utils import canonicalize_name
12from tomlkit.exceptions import TOMLKitError
14from lpip.cache import (
15 CatalogCachePaths,
16 catalog_cache_paths,
17 ensure_catalog_checkout,
18 purge_catalog_cache,
19)
20from lpip.errors import (
21 CatalogAlreadyExistsError,
22 CatalogFileNotFoundError,
23 CatalogSchemaError,
24 DisallowedSchemeError,
25 InvalidPackageNameError,
26 MissingRevisionError,
27 PackageAlreadyExistsError,
28 SubdirectoryEscapeError,
29 TomlParseError,
30 UnknownPackageError,
31 UnknownTomlKeyError,
32 UnsupportedAuthoringUrlError,
33)
34from lpip.models import (
35 EffectiveConfig,
36 PackageSource,
37 PolicyConfig,
38 RefreshMode,
39 SourceConfig,
40 SourceType,
41)
42from lpip.validation import (
43 CATALOG_TOP_KEYS,
44 PACKAGE_FIELD_KEYS,
45 first_unknown_key,
46 normalize_authoring_git_url,
47 optional_str,
48 subdirectory_escapes_repo,
49 url_scheme,
50)
52PACKAGE_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]*$")
53DEFAULT_CATALOG_FILENAME = "lpip-sources.toml"
54CATALOG_INIT_TEMPLATE = """\
55# lpip package catalog
56# Commit this file in the catalog Git repository. Consumers fetch it
57# (source.type = git). Add packages with: lpip catalog add
59[packages]
61# Example:
62# [packages.libfoo]
63# url = "git+ssh://git.company/libs/libfoo.git"
64# revision = "main"
65# subdirectory = "python"
66"""
69@dataclass(frozen=True)
70class Catalog:
71 path: Path
72 packages: tuple[PackageSource, ...]
74 def find(self, name: str) -> PackageSource | None:
75 """Match a distribution name, ignoring PEP 503 normalization."""
76 want = canonicalize_name(name)
77 for pkg in self.packages:
78 if canonicalize_name(pkg.name) == want:
79 return pkg
80 return None
82 def get(self, name: str) -> PackageSource:
83 for pkg in self.packages:
84 if pkg.name == name:
85 return pkg
86 raise UnknownPackageError(name, catalog_path=self.path)
88 def sorted_by_name(self) -> tuple[PackageSource, ...]:
89 return tuple(sorted(self.packages, key=lambda p: p.name))
92def load_catalog(path: Path, policy: PolicyConfig | None = None) -> Catalog:
93 """Parse and validate a catalog file (AC-0003, AC-0010, AC-0011)."""
94 policy = policy or PolicyConfig()
95 catalog_path = path.expanduser()
96 if not catalog_path.is_file():
97 raise CatalogFileNotFoundError(catalog_path)
99 try:
100 with catalog_path.open("rb") as fh:
101 data = tomllib.load(fh)
102 except tomllib.TOMLDecodeError as exc:
103 raise TomlParseError(catalog_path, detail=str(exc)) from exc
105 unknown_top = first_unknown_key(data, CATALOG_TOP_KEYS)
106 if unknown_top is not None:
107 raise UnknownTomlKeyError(unknown_top, path=catalog_path, kind="catalog")
108 if "packages" not in data:
109 raise CatalogSchemaError(
110 "Catalog is missing the packages table.",
111 path=catalog_path,
112 hint="Add a [packages.<name>] table for each installable package.",
113 )
114 packages_raw = data["packages"]
115 if not isinstance(packages_raw, dict): 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true
116 raise CatalogSchemaError(
117 "'packages' must be a TOML table of package names.",
118 path=catalog_path,
119 )
121 loaded: list[PackageSource] = []
122 for name, body in packages_raw.items():
123 loaded.append(_package_from_table(name, body, catalog_path, policy))
124 return Catalog(path=catalog_path, packages=tuple(loaded))
127def _package_from_table(
128 name: str,
129 body: object,
130 catalog_path: Path,
131 policy: PolicyConfig,
132) -> PackageSource:
133 if not isinstance(body, dict): 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 raise CatalogSchemaError(
135 f"Package {name!r} must be a table of fields.",
136 path=catalog_path,
137 )
138 unknown = first_unknown_key(body, PACKAGE_FIELD_KEYS)
139 if unknown is not None:
140 raise UnknownTomlKeyError(
141 f"packages.{name}.{unknown}",
142 path=catalog_path,
143 kind="catalog",
144 )
145 if "url" not in body:
146 raise CatalogSchemaError(
147 f"Package {name!r} is missing required field 'url'.",
148 path=catalog_path,
149 )
150 url = body["url"]
151 if not isinstance(url, str) or not url.strip(): 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true
152 raise CatalogSchemaError(
153 f"Package {name!r} field 'url' must be a non-empty string.",
154 path=catalog_path,
155 )
156 url = url.strip()
157 scheme = url_scheme(url)
158 if scheme not in {s.lower() for s in policy.allowed_schemes}:
159 raise DisallowedSchemeError(
160 url,
161 allowed=policy.allowed_schemes,
162 catalog_path=catalog_path,
163 )
165 try:
166 revision = optional_str(body.get("revision"))
167 subdirectory = optional_str(body.get("subdirectory"))
168 except TypeError as exc:
169 raise CatalogSchemaError(
170 f"Package {name!r} has a non-string optional field: {exc}",
171 path=catalog_path,
172 ) from exc
174 if policy.require_pinned_revision and revision is None:
175 raise MissingRevisionError(name, catalog_path=catalog_path)
177 if subdirectory is not None and subdirectory_escapes_repo(subdirectory):
178 raise SubdirectoryEscapeError(subdirectory, catalog_path=catalog_path)
180 return PackageSource(name=name, url=url, revision=revision, subdirectory=subdirectory)
183def local_catalog_path(cfg: EffectiveConfig, *, cwd: Path) -> Path:
184 return local_path_for_source(cfg.source, cwd=cwd)
187def local_path_for_source(source: SourceConfig, *, cwd: Path) -> Path:
188 raw = Path(source.path).expanduser()
189 return raw if raw.is_absolute() else (cwd / raw)
192@dataclass(frozen=True)
193class ResolvedCatalogLocation:
194 catalog_file: Path
195 cache: CatalogCachePaths | None
198def catalog_file_in_repo(repo: Path, source_path: str) -> Path:
199 if subdirectory_escapes_repo(source_path.replace("\\", "/")): 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 raise SubdirectoryEscapeError(source_path)
201 path = repo / source_path
202 if not path.is_file(): 202 ↛ 203line 202 didn't jump to line 203 because the condition on line 202 was never true
203 raise CatalogFileNotFoundError(path)
204 return path
207def resolve_source_location(
208 source: SourceConfig,
209 *,
210 cwd: Path,
211 cache_home: Path | None = None,
212 force_refresh: bool = False,
213) -> ResolvedCatalogLocation:
214 """Resolve one configured catalog (ADR-0021)."""
215 if source.type is SourceType.FILE:
216 return ResolvedCatalogLocation(
217 catalog_file=local_path_for_source(source, cwd=cwd),
218 cache=None,
219 )
220 if not source.url: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true
221 raise CatalogSchemaError(
222 "source.url is required when source.type is git.",
223 hint="Set source.url in config.toml or LPIP_SOURCE_URL.",
224 )
225 refresh = source.refresh or RefreshMode.AUTO
226 cache = ensure_catalog_checkout(
227 source.url,
228 source.revision,
229 cache_home=cache_home,
230 refresh=refresh,
231 force_refresh=force_refresh,
232 )
233 return ResolvedCatalogLocation(
234 catalog_file=catalog_file_in_repo(cache.repo, source.path),
235 cache=cache,
236 )
239def resolve_catalog_location(
240 cfg: EffectiveConfig,
241 *,
242 cwd: Path,
243 cache_home: Path | None = None,
244 force_refresh: bool = False,
245) -> ResolvedCatalogLocation:
246 """Resolve the first catalog source."""
247 return resolve_source_location(
248 cfg.source,
249 cwd=cwd,
250 cache_home=cache_home,
251 force_refresh=force_refresh,
252 )
255def git_sources(cfg: EffectiveConfig) -> tuple[SourceConfig, ...]:
256 return tuple(src for src in cfg.sources if src.type is SourceType.GIT and src.url)
259def planned_git_cache_paths_for(source: SourceConfig) -> CatalogCachePaths:
260 if source.type is not SourceType.GIT or not source.url: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 raise CatalogSchemaError(
262 "source.type must be git with a url to use the catalog cache.",
263 hint='Set source.type = "git" and source.url.',
264 )
265 return catalog_cache_paths(source.url, source.revision)
268def planned_git_cache_paths(cfg: EffectiveConfig) -> CatalogCachePaths:
269 return planned_git_cache_paths_for(cfg.source)
272def purge_git_catalog_cache(cfg: EffectiveConfig) -> CatalogCachePaths:
273 if cfg.source.type is not SourceType.GIT or not cfg.source.url:
274 raise CatalogSchemaError(
275 "source.type must be git with a url to purge the catalog cache.",
276 hint='Set source.type = "git" and source.url.',
277 )
278 return purge_catalog_cache(cfg.source.url, cfg.source.revision)
281def purge_all_git_catalog_caches(cfg: EffectiveConfig) -> tuple[CatalogCachePaths, ...]:
282 caches = tuple(
283 purge_catalog_cache(src.url, src.revision)
284 for src in git_sources(cfg)
285 if src.url is not None
286 )
287 if not caches: 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true
288 raise CatalogSchemaError(
289 "source.type must be git with a url to purge the catalog cache.",
290 hint='Set source.type = "git" and source.url.',
291 )
292 return caches
295@dataclass(frozen=True)
296class PackageNameClash:
297 """One shadowed package name across catalog sources (AC-0026)."""
299 name: str
300 winner_source: str
301 shadowed_source: str
302 identical: bool
305def _package_definition(package: PackageSource) -> tuple[str, str | None, str | None]:
306 return (package.url, package.revision, package.subdirectory)
309def load_merged_catalog(
310 cfg: EffectiveConfig,
311 *,
312 cwd: Path,
313 cache_home: Path | None = None,
314 force_refresh: bool = False,
315) -> tuple[Catalog, tuple[PackageNameClash, ...]]:
316 """Load every configured source; first package name wins (ADR-0021)."""
317 merged: list[PackageSource] = []
318 first_by_key: dict[str, PackageSource] = {}
319 clashes: list[PackageNameClash] = []
320 first_path: Path | None = None
321 for src in cfg.sources:
322 location = resolve_source_location(
323 src,
324 cwd=cwd,
325 cache_home=cache_home,
326 force_refresh=force_refresh,
327 )
328 if first_path is None:
329 first_path = location.catalog_file
330 catalog = load_catalog(location.catalog_file, cfg.policy)
331 for pkg in catalog.packages:
332 tagged = replace(pkg, catalog_source=src.name)
333 key = canonicalize_name(pkg.name)
334 winner = first_by_key.get(key)
335 if winner is None:
336 first_by_key[key] = tagged
337 merged.append(tagged)
338 continue
339 clashes.append(
340 PackageNameClash(
341 name=winner.name,
342 winner_source=winner.catalog_source or "default",
343 shadowed_source=src.name,
344 identical=_package_definition(winner) == _package_definition(pkg),
345 )
346 )
347 return (
348 Catalog(
349 path=first_path or (cwd / DEFAULT_CATALOG_FILENAME),
350 packages=tuple(merged),
351 ),
352 tuple(clashes),
353 )
356def load_catalog_for_config(
357 cfg: EffectiveConfig,
358 *,
359 cwd: Path,
360 cache_home: Path | None = None,
361 force_refresh: bool = False,
362) -> Catalog:
363 catalog, _clashes = load_merged_catalog(
364 cfg,
365 cwd=cwd,
366 cache_home=cache_home,
367 force_refresh=force_refresh,
368 )
369 return catalog
372def normalize_package_for_catalog(package: PackageSource) -> PackageSource:
373 """Normalize authoring URL; fail if the scheme is not a supported Git form."""
374 stored = normalize_authoring_git_url(package.url)
375 scheme = url_scheme(stored)
376 if scheme not in {"git+ssh", "git+https", "git+file"}: 376 ↛ 377line 376 didn't jump to line 377 because the condition on line 376 was never true
377 raise UnsupportedAuthoringUrlError(package.url)
378 return PackageSource(
379 name=package.name,
380 url=stored,
381 revision=package.revision,
382 subdirectory=package.subdirectory,
383 )
386def _validate_new_package(package: PackageSource, policy: PolicyConfig, catalog_path: Path) -> None:
387 if not PACKAGE_NAME_RE.fullmatch(package.name): 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true
388 raise InvalidPackageNameError(package.name)
389 _package_from_table(
390 package.name,
391 {
392 "url": package.url,
393 **({"revision": package.revision} if package.revision else {}),
394 **({"subdirectory": package.subdirectory} if package.subdirectory else {}),
395 },
396 catalog_path,
397 policy,
398 )
401def resolve_authoring_catalog_path(path: str | Path | None, *, cwd: Path) -> Path:
402 """Working-tree catalog path (ADR-0015). Does not use consumer source.type."""
403 raw = Path(path) if path is not None else Path(DEFAULT_CATALOG_FILENAME)
404 raw = raw.expanduser()
405 return raw if raw.is_absolute() else (cwd / raw)
408def write_catalog_init_template(path: Path) -> Path:
409 """Write a commented empty catalog. Fails if the path already exists."""
410 if path.exists():
411 raise CatalogAlreadyExistsError(path)
412 path.parent.mkdir(parents=True, exist_ok=True)
413 path.write_text(CATALOG_INIT_TEMPLATE, encoding="utf-8")
414 return path
417def add_package_to_catalog_file(
418 path: Path,
419 package: PackageSource,
420 *,
421 policy: PolicyConfig | None = None,
422 force: bool = False,
423) -> Path:
424 """Append or replace a package table in a working-tree catalog (ADR-0015)."""
425 path = path.expanduser()
426 policy = policy or PolicyConfig()
427 package = normalize_package_for_catalog(package)
428 _validate_new_package(package, policy, path)
430 if path.is_file():
431 try:
432 document = tomlkit.parse(path.read_text(encoding="utf-8"))
433 except TOMLKitError as exc:
434 raise TomlParseError(path, detail=str(exc)) from exc
435 else:
436 path.parent.mkdir(parents=True, exist_ok=True)
437 document = tomlkit.document()
438 document.add(tomlkit.comment("lpip package catalog"))
440 packages = document.get("packages")
441 if packages is None:
442 packages = tomlkit.table()
443 document["packages"] = packages
444 elif not isinstance(packages, dict): 444 ↛ 445line 444 didn't jump to line 445 because the condition on line 444 was never true
445 raise CatalogSchemaError(
446 "'packages' must be a TOML table of package names.",
447 path=path,
448 )
449 if package.name in packages and not force:
450 raise PackageAlreadyExistsError(package.name, catalog_path=path)
452 entry = tomlkit.table()
453 entry["url"] = package.url
454 if package.revision:
455 entry["revision"] = package.revision
456 if package.subdirectory:
457 entry["subdirectory"] = package.subdirectory
458 packages[package.name] = entry
459 path.write_text(tomlkit.dumps(document), encoding="utf-8")
460 return path