Coverage for webapp/publisher/logic.py: 98%

1from typing import TypedDict, List, Union, Dict 

2from collections import Counter 

3from datetime import datetime 

4from webapp.observability.utils import trace_function 

5from webapp.store.logic import process_revision 

6 

7Error = TypedDict("Error", {"code": str, "message": str}) 

8 

9Base = TypedDict("Base", {"architecture": str, "channel": str, "name": str}) 

10 

11Revision = TypedDict( 

12 "Revision", 

13 { 

14 "bases": List[Base], 

15 "created-at": str, 

16 "revision": int, 

17 "sha3-384": str, 

18 "size": int, 

19 "status": str, 

20 "version": str, 

21 "errors": Union[List[Error], None], 

22 }, 

23) 

24 

25Resource = TypedDict( 

26 "Resource", {"name": str, "revision": Union[int, None], "type": str} 

27) 

28 

29Release = TypedDict( 

30 "Release", {"revision": Revision, "resources": List[Resource]} 

31) 

32 

33ReleaseMap = TypedDict( 

34 "ReleaseMap", 

35 { 

36 "track": str, 

37 "risk": str, 

38 "releases": List[Release], 

39 }, 

40) 

41 

42 

43@trace_function 

44def process_releases( 

45 channel_map, channels, revisions: List[Revision] 

46) -> Dict[str, ReleaseMap]: 

47 """ 

48 Process the releases and return a dictionary with the 

49 channel name as key and a list of releases as value. 

50 

51 Args: 

52 channel_map: Mapping of revisions to channels and resources 

53 channels: List of channels 

54 revisions: List of revisions 

55 

56 Returns: 

57 Dictionary with channel name as key and a list of releases as value 

58 """ 

59 

60 res = {} 

61 for channel in channels: 

62 res[channel["name"]] = {} 

63 res[channel["name"]]["track"] = channel["track"] 

64 res[channel["name"]]["risk"] = channel["risk"] 

65 res[channel["name"]]["releases"] = {} 

66 

67 revision_map = {} 

68 

69 for revision in revisions: 

70 revision_map[revision["revision"]] = process_revision(revision) 

71 

72 for channel in channel_map: 

73 revision = revision_map[channel["revision"]] 

74 resources = channel["resources"] 

75 res[channel["channel"]]["releases"][revision["revision"]] = { 

76 "revision": revision, 

77 "resources": resources, 

78 } 

79 

80 for channel in list(res): 

81 releases = list(res[channel]["releases"].values()) 

82 

83 releases.sort( 

84 key=lambda x: datetime.strptime( 

85 x["revision"]["created-at"], "%Y-%m-%dT%H:%M:%SZ" 

86 ), 

87 reverse=True, 

88 ) 

89 

90 if len(releases) > 0: 

91 res[channel]["releases"] = releases 

92 else: 

93 del res[channel] 

94 

95 return res 

96 

97 

98@trace_function 

99def get_all_architectures(releases: Dict[str, ReleaseMap]) -> List[str]: 

100 """ 

101 Get all architectures from the releases 

102 

103 Args: 

104 releases: Dictionary with channel name as key 

105 and a list of releases as value 

106 

107 Returns: 

108 List of architectures, sorted by frequency 

109 

110 """ 

111 architectures = Counter() 

112 for channel in releases: 

113 for release in releases[channel]["releases"]: 

114 for base in release["revision"]["bases"]: 

115 architectures.update([base["architecture"]]) 

116 

117 return [arch for arch, _ in architectures.most_common()]