Coverage for daklib/ls.py: 63%

1"""List packages according to various criteria 

2 

3@copyright: 2014, Ansgar Burchardt <ansgar@debian.org> 

4@license: GPL-2+ 

5""" 

6 

7# This program is free software; you can redistribute it and/or modify 

8# it under the terms of the GNU General Public License as published by 

9# the Free Software Foundation; either version 2 of the License, or 

10# (at your option) any later version. 

11# 

12# This program is distributed in the hope that it will be useful, 

13# but WITHOUT ANY WARRANTY; without even the implied warranty of 

14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

15# GNU General Public License for more details. 

16# 

17# You should have received a copy of the GNU General Public License 

18# along with this program; if not, write to the Free Software 

19# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 

20 

21from collections import defaultdict 

22from collections.abc import Collection, Iterable 

23 

24import sqlalchemy.dialects.postgresql as pgsql 

25from sqlalchemy import ColumnElement, sql 

26 

27from daklib.dbconn import DBConn, package_list 

28 

29 

30def list_packages( 

31 packages: Collection[str], 

32 suites: Collection[str] | None = None, 

33 components: Collection[str] | None = None, 

34 architectures: Collection[str] | None = None, 

35 binary_types: Collection[str] | None = None, 

36 source_and_binary=False, 

37 regex=False, 

38 format: str | None = None, 

39 highest: str | None = None, 

40) -> Iterable[str] | Iterable[dict[str, dict[str, dict[str, dict[str, str]]]]]: 

41 session = DBConn().session() 

42 try: 

43 t = package_list 

44 

45 comparison_operator = "~" if regex else "=" 

46 

47 where: ColumnElement[bool] = sql.false() 

48 for package in packages: 

49 where = where | t.c.package.op(comparison_operator)(package) 

50 if source_and_binary: 

51 where = where | t.c.source.op(comparison_operator)(package) 

52 

53 if suites is not None: 

54 where = where & (t.c.suite.in_(suites) | t.c.codename.in_(suites)) 

55 if components is not None: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true

56 where = where & t.c.component.in_(components) 

57 if architectures is not None: 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true

58 where = where & t.c.architecture.in_(architectures) 

59 if binary_types is not None: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 where = where & t.c.type.in_(binary_types) 

61 

62 if format is None: 

63 c_architectures = sql.func.string_agg( 

64 t.c.architecture, 

65 pgsql.aggregate_order_by( 

66 sql.literal(", "), 

67 t.c.architecture_is_source.desc(), 

68 t.c.architecture, 

69 ), 

70 ) 

71 query = ( 

72 sql.select(t.c.package, t.c.version, t.c.display_suite, c_architectures) 

73 .where(where) 

74 .group_by(t.c.package, t.c.version, t.c.display_suite) 

75 .order_by(t.c.package, t.c.version, t.c.display_suite) 

76 ) 

77 result = session.execute(query).mappings().all() 

78 

79 if len(result) == 0: 

80 return 

81 

82 lengths = { 

83 "package": max(10, max(len(row[t.c.package]) for row in result)), 

84 "version": max(13, max(len(row[t.c.version]) for row in result)), 

85 "suite": max(10, max(len(row[t.c.display_suite]) for row in result)), 

86 } 

87 format = "{0:{lengths[package]}} | {1:{lengths[version]}} | {2:{lengths[suite]}} | {3}" 

88 

89 for row in result: 

90 yield format.format( 

91 row[t.c.package], 

92 row[t.c.version], 

93 row[t.c.display_suite], 

94 row[c_architectures], 

95 lengths=lengths, 

96 ) 

97 elif format in ("control-suite", "heidi"): 97 ↛ 103line 97 didn't jump to line 103 because the condition on line 97 was always true

98 query = sql.select(t.c.package, t.c.version, t.c.architecture).where(where) 

99 for row in session.execute(query).mappings(): 

100 yield "{0} {1} {2}".format( 

101 row[t.c.package], row[t.c.version], row[t.c.architecture] 

102 ) 

103 elif format == "python": 

104 c_architectures = sql.func.string_agg( 

105 t.c.architecture, 

106 pgsql.aggregate_order_by( 

107 sql.literal(","), 

108 t.c.architecture_is_source.desc(), 

109 t.c.architecture, 

110 ), 

111 ) 

112 query = ( 

113 sql.select( 

114 t.c.package, 

115 t.c.version, 

116 t.c.display_suite, 

117 c_architectures, 

118 t.c.source, 

119 t.c.source_version, 

120 t.c.component, 

121 ) 

122 .where(where) 

123 .group_by( 

124 t.c.package, 

125 t.c.version, 

126 t.c.display_suite, 

127 t.c.source, 

128 t.c.component, 

129 t.c.source_version, 

130 ) 

131 ) 

132 result = session.execute(query).mappings().all() 

133 

134 if len(result) == 0: 

135 return 

136 

137 def val(): 

138 return defaultdict(val) 

139 

140 ret = val() 

141 for row in result: 

142 ret[row[t.c.package]][row[t.c.display_suite]][row[t.c.version]] = { 

143 "component": row[t.c.component], 

144 "architectures": row[c_architectures].split(","), 

145 "source": row[t.c.source], 

146 "source_version": row[t.c.source_version], 

147 } 

148 

149 yield ret 

150 return 

151 else: 

152 raise ValueError("Unknown output format requested.") 

153 

154 if highest is not None: 154 ↛ 155line 154 didn't jump to line 155

155 query = ( 

156 sql.select(t.c.package, sql.func.max(t.c.version)) 

157 .where(where) 

158 .group_by(t.c.package) 

159 .order_by(t.c.package) 

160 ) 

161 yield "" 

162 for r in session.execute(query): 

163 yield "{0} ({1} {2})".format(r[0], highest, r[1]) 

164 finally: 

165 session.close()