Coverage for src/secchi/workflows/compare.py: 29%

1"""Package comparison workflow.""" 

2 

3from secchi.errors import PackageNotFoundError, SecchiError 

4from secchi.models import PackageRef, Registry 

5from secchi.services.comparison import ComparisonResult, compare_intelligence 

6from secchi.services.intelligence import PackageIntelligenceService 

7from secchi.services.resolver import parse_package_spec, resolve_package 

8 

9 

10async def _resolve_refs(specs: list[str], registry: str | None) -> list[PackageRef]: 

11 refs: list[PackageRef] = [] 

12 preference = {item: index for index, item in enumerate(Registry)} 

13 for spec in specs: 

14 if registry or ":" in spec: 

15 refs.append(parse_package_spec(spec, None if ":" in spec else registry)) 

16 continue 

17 matches = await resolve_package(spec) 

18 if not matches: 

19 raise PackageNotFoundError( 

20 f"No exact package named '{spec}' was found across registries." 

21 ) 

22 refs.append(sorted(matches, key=lambda ref: preference[ref.registry])[0]) 

23 return refs 

24 

25 

26async def run( 

27 specs: list[str], 

28 *, 

29 registry: str | None = None, 

30 refresh: bool = False, 

31 service: PackageIntelligenceService | None = None, 

32) -> ComparisonResult: 

33 if len(specs) < 2: 

34 raise SecchiError("Compare requires at least two packages.") 

35 refs = await _resolve_refs(specs, registry) 

36 pipeline = service or PackageIntelligenceService() 

37 intelligence = await pipeline.fetch_project(refs, force_refresh=refresh) 

38 return compare_intelligence(list(intelligence.results.values()))