Coverage for dakweb/routers/changelog.py: 94%

1# SPDX-License-Identifier: GPL-2.0-or-later 

2# © 2025, Sérgio Cipriano <cipriano@debian.org> 

3# © 2026, Ansgar 🙀 <ansgar@debian.org> 

4# © 2026, Anton Gladky <gladk@debian.org> 

5 

6from datetime import datetime 

7from typing import Annotated 

8 

9from fastapi import APIRouter, Depends, Query, Response 

10from pydantic import AfterValidator, BaseModel 

11from sqlalchemy import TIMESTAMP, cast, or_, select 

12from sqlalchemy.orm import Session 

13 

14from daklib.dbconn import DBChange, DBChangelog 

15from dakweb.routers.deps import get_db 

16 

17router = APIRouter() 

18 

19 

20class ChangelogV1(BaseModel): 

21 date: str 

22 source: str 

23 version: str 

24 changedby: str 

25 changelog: str 

26 

27 

28@router.get("/changelogs") 

29def changelogs( 

30 response: Response, 

31 search_term: Annotated[ 

32 str, 

33 AfterValidator(str.strip), 

34 Query( 

35 description="Substring to match in changelog text or changedby", 

36 min_length=1, 

37 ), 

38 ], 

39 db: Session = Depends(get_db), 

40) -> list[ChangelogV1]: 

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

42 

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

44 """ 

45 

46 # Add advisory / deprecation style headers 

47 response.headers["Warning"] = ( 

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

49 ) 

50 response.headers["X-Legacy-Endpoint"] = "true" 

51 response.headers["X-Preferred-Endpoint"] = "/v2/changelogs" 

52 

53 stmt = ( 

54 select( 

55 DBChange.date, 

56 DBChange.source, 

57 DBChange.version, 

58 DBChange.changedby, 

59 DBChangelog.changelog, 

60 ) 

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

62 .where(DBChange.source != "debian-keyring") 

63 .where( 

64 or_( 

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

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

67 ) 

68 ) 

69 .order_by(DBChange.seen) 

70 ) 

71 

72 return [ 

73 ChangelogV1( 

74 date=c.date, 

75 source=c.source, 

76 version=c.version, 

77 changedby=c.changedby, 

78 changelog=c.changelog, 

79 ) 

80 for c in db.execute(stmt) 

81 ] 

82 

83 

84class Changelog(BaseModel): 

85 id: int 

86 date: str 

87 source: str 

88 version: str 

89 changedby: str 

90 changelog: str 

91 

92 

93class ChangelogResponse(BaseModel): 

94 count: int 

95 limit: int 

96 offset: int 

97 has_more: bool 

98 results: list[Changelog] 

99 

100 

101@router.get("/v2/changelogs") 

102def changelogs_v2( 

103 search_term: Annotated[ 

104 Annotated[str, AfterValidator(str.strip)] | None, 

105 Query(description="Substring to match in changelog text or changedby"), 

106 ] = None, 

107 source: Annotated[ 

108 Annotated[str, AfterValidator(str.strip)] | None, 

109 Query(description="Exact source package name to restrict"), 

110 ] = None, 

111 since: Annotated[datetime | None, Query(description="ISO date/datetime")] = None, 

112 till: Annotated[ 

113 datetime | None, Query(description="ISO date/datetime upper bound (inclusive)") 

114 ] = None, 

115 since_id: Annotated[ 

116 int | None, Query(description="Return only rows with id > since_id", ge=0) 

117 ] = None, 

118 limit: Annotated[int, Query(ge=1, le=500)] = 50, 

119 offset: Annotated[int, Query(ge=0)] = 0, 

120 db: Session = Depends(get_db), 

121) -> ChangelogResponse: 

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

123 

124 stmt = ( 

125 select( 

126 DBChange.change_id, 

127 DBChange.date, 

128 DBChange.source, 

129 DBChange.version, 

130 DBChange.changedby, 

131 DBChangelog.changelog, 

132 ) 

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

134 .where(DBChange.source != "debian-keyring") 

135 ) 

136 

137 if search_term: 

138 like = f"%{search_term}%" 

139 stmt = stmt.where( 

140 or_( 

141 DBChangelog.changelog.ilike(like), 

142 DBChange.changedby.ilike(like), 

143 ) 

144 ) 

145 

146 if source: 

147 stmt = stmt.where(DBChange.source == source) 

148 

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

150 if since: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true

151 stmt = stmt.where(cast(DBChange.date, TIMESTAMP) >= since) 

152 if till: 152 ↛ 154line 152 didn't jump to line 154 because the condition on line 152 was never true

153 # Inclusive upper bound 

154 stmt = stmt.where(cast(DBChange.date, TIMESTAMP) <= till) 

155 if since_id is not None: 

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

157 stmt = stmt.where(DBChange.change_id > since_id) 

158 

159 # Order by change_id so since_id acts as a stable cursor for pagination. 

160 stmt = stmt.order_by(DBChange.change_id) 

161 

162 rows = db.execute(stmt.offset(offset).limit(limit + 1)).all() 

163 has_more = len(rows) > limit 

164 rows = rows[:limit] 

165 

166 results = [ 

167 Changelog( 

168 id=r.change_id, 

169 date=r.date, 

170 source=r.source, 

171 version=r.version, 

172 changedby=r.changedby, 

173 changelog=r.changelog, 

174 ) 

175 for r in rows 

176 ] 

177 return ChangelogResponse( 

178 count=len(results), 

179 limit=limit, 

180 offset=offset, 

181 has_more=has_more, 

182 results=results, 

183 )