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

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 security_refresh: bool = False, 

32 service: PackageIntelligenceService | None = None, 

33) -> ComparisonResult: 

34 if len(specs) < 2: 

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

36 refs = await _resolve_refs(specs, registry) 

37 pipeline = service or PackageIntelligenceService() 

38 if security_refresh: 

39 intelligence = await pipeline.fetch_project( 

40 refs, force_refresh=refresh, force_security_refresh=True 

41 ) 

42 else: 

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

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