Coverage for src/secchi/workflows/common.py: 100%

1"""Shared workflow primitives and application-level errors.""" 

2 

3from __future__ import annotations 

4 

5from secchi.errors import PackageNotFoundError, RegistryUnavailableError, SecchiError 

6from secchi.models import PackageRef 

7from secchi.services.intelligence import IntelligenceResult, PackageIntelligenceService 

8 

9 

10class WorkflowError(SecchiError): 

11 """A user-actionable workflow failure suitable for CLI/MCP mapping.""" 

12 

13 

14async def fetch_package( 

15 ref: PackageRef, 

16 refresh: bool = False, 

17 *, 

18 service: PackageIntelligenceService | None = None, 

19) -> IntelligenceResult: 

20 pipeline = service or PackageIntelligenceService() 

21 return await pipeline.fetch_package(ref, force_refresh=refresh) 

22 

23 

24async def require_package( 

25 ref: PackageRef, 

26 refresh: bool = False, 

27 *, 

28 service: PackageIntelligenceService | None = None, 

29) -> IntelligenceResult: 

30 result = await fetch_package(ref, refresh, service=service) 

31 if result.error or result.info is None or result.derived is None: 

32 message = result.error.message if result.error else "No package data returned." 

33 error_type = RegistryUnavailableError if result.error else PackageNotFoundError 

34 raise error_type( 

35 f"Could not load {ref.name} from {ref.registry.value}: {message}" 

36 ) from None 

37 return result