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 - limit: (optional) default 50, max 500 

92 - offset: (optional) default 0 

93 

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

95 """ 

96 

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

98 

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

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

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

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

103 

104 # Pagination params 

105 try: 

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

107 except ValueError: 

108 limit = 50 

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

110 try: 

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

112 except ValueError: 

113 offset = 0 

114 offset = max(0, offset) 

115 

116 def parse_dt(val: str): 

117 if not val: 

118 return None 

119 # Accept date-only or full isoformat 

120 try: 

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

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

123 return datetime.fromisoformat(val) 

124 except Exception: 

125 return None 

126 

127 since_dt = parse_dt(since_param) 

128 till_dt = parse_dt(till_param) 

129 

130 if since_param and not since_dt: 

131 bottle.response.status = 400 

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

133 if till_param and not till_dt: 

134 bottle.response.status = 400 

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

136 

137 s = DBConn().session() 

138 try: 

139 q = ( 

140 s.query( 

141 DBChange.change_id, 

142 DBChange.date, 

143 DBChange.source, 

144 DBChange.version, 

145 DBChange.changedby, 

146 DBChangelog.changelog, 

147 ) 

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

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

150 ) 

151 

152 if search_term: 

153 like = f"%{search_term}%" 

154 q = q.filter( 

155 or_( 

156 DBChangelog.changelog.ilike(like), 

157 DBChange.changedby.ilike(like), 

158 ) 

159 ) 

160 if source_filter: 

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

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

163 if since_dt: 

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

165 if till_dt: 

166 # Inclusive upper bound 

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

168 

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

170 

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

172 has_more = len(rows) > limit 

173 rows = rows[:limit] 

174 

175 results = [ 

176 { 

177 "id": r.change_id, 

178 "date": ( 

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

180 ), 

181 "source": r.source, 

182 "version": r.version, 

183 "changedby": r.changedby, 

184 "changelog": r.changelog, 

185 } 

186 for r in rows 

187 ] 

188 

189 return json.dumps( 

190 { 

191 "count": len(results), 

192 "limit": limit, 

193 "offset": offset, 

194 "has_more": has_more, 

195 "results": results, 

196 } 

197 ) 

198 finally: 

199 s.close() 

200 

201 

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