Coverage for dakweb/queries/changelog.py: 0%

1"""Packages' changelogs related queries 

2 

3@contact: Debian FTPMaster <ftpmaster@debian.org> 

4@copyright: 2025 Sérgio Cipriano <cipriano@debian.org> 

5@license: GNU General Public License version 2 or later 

6""" 

7 

8import json 

9from datetime import datetime 

10 

11import bottle 

12from sqlalchemy import TIMESTAMP, cast, or_ 

13 

14from daklib.dbconn import DBChange, DBChangelog, DBConn 

15from dakweb.webregister import QueryRegister 

16 

17 

18@bottle.route("/changelogs") 

19def changelogs() -> str: 

20 """Legacy endpoint; prefer /v2/changelogs. 

21 

22 Returns all matching rows (no pagination). Use only for backward compatibility. 

23 Required param: search_term. 

24 """ 

25 

26 search_term = bottle.request.query.get("search_term", "").strip() 

27 

28 bottle.response.content_type = "application/json; charset=UTF-8" 

29 # Add advisory / deprecation style headers 

30 bottle.response.set_header( 

31 "Warning", 

32 '299 dakweb "/changelogs is legacy; prefer /v2/changelogs (adds pagination & filters)"', 

33 ) 

34 bottle.response.set_header("X-Legacy-Endpoint", "true") 

35 bottle.response.set_header("X-Preferred-Endpoint", "/v2/changelogs") 

36 

37 if not search_term: 

38 bottle.response.status = 400 

39 return json.dumps({"error": "Missing required query parameter: search_term"}) 

40 

41 s = DBConn().session() 

42 

43 q = ( 

44 s.query( 

45 DBChange.date, 

46 DBChange.source, 

47 DBChange.version, 

48 DBChange.changedby, 

49 DBChangelog.changelog, 

50 ) 

51 .join(DBChangelog, DBChange.changelog_id == DBChangelog.id) 

52 .filter(DBChange.source != "debian-keyring") 

53 .filter( 

54 or_( 

55 DBChangelog.changelog.ilike(f"%{search_term}%"), 

56 DBChange.changedby.ilike(f"%{search_term}%"), 

57 ) 

58 ) 

59 .order_by(DBChange.seen) 

60 ) 

61 

62 ret = [] 

63 for c in q: 

64 ret.append( 

65 { 

66 "date": c.date, 

67 "source": c.source, 

68 "version": c.version, 

69 "changedby": c.changedby, 

70 "changelog": c.changelog, 

71 } 

72 ) 

73 

74 s.close() 

75 

76 return json.dumps(ret) 

77 

78 

79QueryRegister().register_path("/changelogs", changelogs) 

80 

81 

82@bottle.route("/v2/changelogs") 

83def changelogs_v2() -> str: 

84 """Improved changelogs endpoint with date range & pagination. 

85 

86 Query parameters: 

87 - search_term: (optional) substring to match in changelog text or changedby (ILIKE) 

88 - since: (optional) ISO date/datetime (e.g. 2024-01-01 or 2024-01-01T12:00:00) 

89 - till: (optional) ISO date/datetime upper bound (inclusive) 

90 - source: (optional) exact source package name to restrict 

91 - since_id: (optional) return only rows with id > since_id (id-based polling) 

92 - limit: (optional) default 50, max 500 

93 - offset: (optional) default 0 

94 

95 Returns JSON object with keys: count, limit, offset, has_more, results[]. 

96 """ 

97 

98 bottle.response.content_type = "application/json; charset=UTF-8" 

99 

100 search_term = bottle.request.query.get("search_term", "").strip() 

101 source_filter = bottle.request.query.get("source", "").strip() 

102 since_param = bottle.request.query.get("since", "").strip() 

103 till_param = bottle.request.query.get("till", "").strip() 

104 since_id_str = bottle.request.query.get("since_id", "").strip() 

105 

106 # Pagination params 

107 try: 

108 limit = int(bottle.request.query.get("limit", "50")) 

109 except ValueError: 

110 limit = 50 

111 limit = max(1, min(limit, 500)) 

112 try: 

113 offset = int(bottle.request.query.get("offset", "0")) 

114 except ValueError: 

115 offset = 0 

116 offset = max(0, offset) 

117 

118 def parse_dt(val: str): 

119 if not val: 

120 return None 

121 # Accept date-only or full isoformat 

122 try: 

123 if len(val) == 10: # YYYY-MM-DD 

124 return datetime.strptime(val, "%Y-%m-%d") 

125 return datetime.fromisoformat(val) 

126 except Exception: 

127 return None 

128 

129 since_dt = parse_dt(since_param) 

130 till_dt = parse_dt(till_param) 

131 

132 if since_param and not since_dt: 

133 bottle.response.status = 400 

134 return json.dumps({"error": "Invalid since parameter (expected ISO date)"}) 

135 if till_param and not till_dt: 

136 bottle.response.status = 400 

137 return json.dumps({"error": "Invalid till parameter (expected ISO date)"}) 

138 

139 # Validate id-based filter 

140 since_id = None 

141 if since_id_str: 

142 try: 

143 since_id = int(since_id_str) 

144 except Exception: 

145 bottle.response.status = 400 

146 return json.dumps( 

147 {"error": "Invalid since_id parameter (expected integer)"} 

148 ) 

149 if since_id < 0: 

150 bottle.response.status = 400 

151 return json.dumps({"error": "since_id must be >= 0"}) 

152 

153 s = DBConn().session() 

154 try: 

155 q = ( 

156 s.query( 

157 DBChange.change_id, 

158 DBChange.date, 

159 DBChange.source, 

160 DBChange.version, 

161 DBChange.changedby, 

162 DBChangelog.changelog, 

163 ) 

164 .join(DBChangelog, DBChange.changelog_id == DBChangelog.id) 

165 .filter(DBChange.source != "debian-keyring") 

166 ) 

167 

168 if search_term: 

169 like = f"%{search_term}%" 

170 q = q.filter( 

171 or_( 

172 DBChangelog.changelog.ilike(like), 

173 DBChange.changedby.ilike(like), 

174 ) 

175 ) 

176 if source_filter: 

177 q = q.filter(DBChange.source == source_filter) 

178 # Date column may be stored as TEXT; cast to timestamp for comparison 

179 if since_dt: 

180 q = q.filter(cast(DBChange.date, TIMESTAMP) >= since_dt) 

181 if till_dt: 

182 # Inclusive upper bound 

183 q = q.filter(cast(DBChange.date, TIMESTAMP) <= till_dt) 

184 if since_id is not None: 

185 # Strictly greater than to avoid returning the last seen row again 

186 q = q.filter(DBChange.change_id > since_id) 

187 

188 q = q.order_by(DBChange.seen, DBChange.change_id) # deterministic ordering 

189 

190 rows = q.offset(offset).limit(limit + 1).all() # fetch one extra for has_more 

191 has_more = len(rows) > limit 

192 rows = rows[:limit] 

193 

194 results = [ 

195 { 

196 "id": r.change_id, 

197 "date": ( 

198 r.date.isoformat() if hasattr(r.date, "isoformat") else str(r.date) 

199 ), 

200 "source": r.source, 

201 "version": r.version, 

202 "changedby": r.changedby, 

203 "changelog": r.changelog, 

204 } 

205 for r in rows 

206 ] 

207 

208 return json.dumps( 

209 { 

210 "count": len(results), 

211 "limit": limit, 

212 "offset": offset, 

213 "has_more": has_more, 

214 "results": results, 

215 } 

216 ) 

217 finally: 

218 s.close() 

219 

220 

221QueryRegister().register_path("/v2/changelogs", changelogs_v2)