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

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

19 service: PackageIntelligenceService | None = None, 

20) -> IntelligenceResult: 

21 pipeline = service or PackageIntelligenceService() 

22 if security_refresh: 

23 return await pipeline.fetch_package( 

24 ref, force_refresh=refresh, force_security_refresh=True 

25 ) 

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

27 

28 

29async def require_package( 

30 ref: PackageRef, 

31 refresh: bool = False, 

32 *, 

33 security_refresh: bool = False, 

34 service: PackageIntelligenceService | None = None, 

35) -> IntelligenceResult: 

36 result = await fetch_package( 

37 ref, refresh, security_refresh=security_refresh, service=service 

38 ) 

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

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

41 error_type = RegistryUnavailableError if result.error else PackageNotFoundError 

42 raise error_type( 

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

44 ) from None 

45 return result