Coverage for src/secchi/services/resolver.py: 79%

1"""Resolve direct CLI package names into registry package references.""" 

2 

3from __future__ import annotations 

4 

5from secchi.diagnostics import DiagnosticLog 

6from secchi.errors import ConfigError 

7from secchi.models import PackageRef, Registry 

8from secchi.services.search import PackageSearchService 

9 

10 

11def parse_package_spec(spec: str, registry: str | None = None) -> PackageRef: 

12 """Resolve ``name`` or ``registry:name`` with PyPI as the safe default.""" 

13 if ":" in spec and registry is None: 

14 candidate_registry, name = spec.split(":", 1) 

15 try: 

16 return PackageRef(name=name, registry=Registry(candidate_registry)) 

17 except ValueError: 

18 pass 

19 try: 

20 selected = Registry(registry) if registry else Registry.PYPI 

21 except ValueError as exc: 

22 supported = ", ".join(item.value for item in Registry) 

23 raise ConfigError( 

24 f"Unknown registry '{registry}'. Supported: {supported}" 

25 ) from exc 

26 return PackageRef(name=spec, registry=selected) 

27 

28 

29async def resolve_package( 

30 spec: str, 

31 *, 

32 configured_refs: list[PackageRef] | None = None, 

33 registry: str | None = None, 

34 diagnostics: DiagnosticLog | None = None, 

35) -> list[PackageRef]: 

36 """Resolve a package spec from config first, then exact registry matches.""" 

37 explicit = parse_package_spec(spec, registry) 

38 if registry is not None or ":" in spec: 

39 return [explicit] 

40 

41 configured = [ 

42 ref for ref in configured_refs or [] if ref.name.casefold() == spec.casefold() 

43 ] 

44 if configured: 

45 return configured 

46 

47 matches = await PackageSearchService(diagnostics=diagnostics).search(spec) 

48 exact = [result for result in matches if result.exact] 

49 return [PackageRef(result.name, result.registry) for result in exact]