Coverage for dakweb/routers/source.py: 100%

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

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

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

4 

5from fastapi import APIRouter, Depends 

6from pydantic import BaseModel 

7from sqlalchemy import or_, select 

8from sqlalchemy.orm import Session 

9 

10from daklib.dbconn import ( 

11 DBSource, 

12 DSCFile, 

13 MetadataKey, 

14 PoolFile, 

15 SourceMetadata, 

16 Suite, 

17) 

18from dakweb.routers.deps import get_db 

19 

20router = APIRouter() 

21 

22 

23class DscInSuite(BaseModel): 

24 version: str 

25 component: str 

26 filename: str 

27 filesize: int 

28 sha256sum: str 

29 

30 

31@router.get("/dsc_in_suite/{suite}/{source}") 

32def dsc_in_suite( 

33 suite: str, 

34 source: str, 

35 db: Session = Depends(get_db), 

36) -> list[DscInSuite]: 

37 """Find all dsc files for a given source package in a given suite.""" 

38 

39 stmt = ( 

40 select(DSCFile) 

41 .join(PoolFile) 

42 .join(DBSource) 

43 .join(Suite, DBSource.suites) 

44 .where(or_(Suite.suite_name == suite, Suite.codename == suite)) 

45 .where(DBSource.source == source) 

46 .where(PoolFile.filename.endswith(".dsc")) 

47 ) 

48 

49 return [ 

50 DscInSuite( 

51 version=p.source.version, 

52 component=p.poolfile.component.component_name, 

53 filename=p.poolfile.filename, 

54 filesize=p.poolfile.filesize, 

55 sha256sum=p.poolfile.sha256sum, 

56 ) 

57 for p in db.scalars(stmt) 

58 ] 

59 

60 

61class FileInArchive(BaseModel): 

62 filename: str 

63 component: str 

64 sha256sum: str 

65 

66 

67@router.get("/file_in_archive/{filepattern:path}") 

68def file_in_archive( 

69 filepattern: str, 

70 db: Session = Depends(get_db), 

71) -> list[FileInArchive]: 

72 """Check if a file pattern is known to the archive (SQL LIKE syntax).""" 

73 

74 stmt = select(PoolFile).where(PoolFile.filename.like(filepattern)) 

75 

76 return [ 

77 FileInArchive( 

78 filename=p.filename, 

79 component=p.component.component_name, 

80 sha256sum=p.sha256sum, 

81 ) 

82 for p in db.scalars(stmt) 

83 ] 

84 

85 

86@router.get("/sha256sum_in_archive/{sha256sum}") 

87def sha256sum_in_archive( 

88 sha256sum: str, 

89 db: Session = Depends(get_db), 

90) -> list[FileInArchive]: 

91 """Check if files with matching sha256sums are known to the archive.""" 

92 

93 stmt = select(PoolFile).where(PoolFile.sha256sum == sha256sum) 

94 

95 return [ 

96 FileInArchive( 

97 filename=p.filename, 

98 component=p.component.component_name, 

99 sha256sum=p.sha256sum, 

100 ) 

101 for p in db.scalars(stmt) 

102 ] 

103 

104 

105class SourcesInSuite(BaseModel): 

106 source: str 

107 version: str 

108 

109 

110@router.get("/sources_in_suite/{suite}") 

111def sources_in_suite( 

112 suite: str, 

113 db: Session = Depends(get_db), 

114) -> list[SourcesInSuite]: 

115 """Return all source packages and versions in a given suite.""" 

116 

117 stmt = ( 

118 select(DBSource) 

119 .join(Suite, DBSource.suites) 

120 .where(or_(Suite.suite_name == suite, Suite.codename == suite)) 

121 ) 

122 

123 return [ 

124 SourcesInSuite(source=p.source, version=p.version) for p in db.scalars(stmt) 

125 ] 

126 

127 

128@router.get("/all_sources") 

129def all_sources(db: Session = Depends(get_db)) -> list[SourcesInSuite]: 

130 """Return all source packages and versions known to the archive (includes NEW).""" 

131 

132 return [ 

133 SourcesInSuite(source=p.source, version=p.version) 

134 for p in db.scalars(select(DBSource)) 

135 ] 

136 

137 

138class SourceByMetadata(BaseModel): 

139 source: str 

140 metadata_value: str 

141 

142 

143@router.get("/source/by_metadata/{key}") 

144def source_by_metadata( 

145 key: str, 

146 db: Session = Depends(get_db), 

147) -> list[SourceByMetadata]: 

148 """Find all Debian source packages which have the specified metadata set.""" 

149 

150 stmt = ( 

151 select(DBSource.source, SourceMetadata.value) 

152 .join(SourceMetadata) 

153 .join(MetadataKey) 

154 .where(MetadataKey.key == key) 

155 ) 

156 

157 return [ 

158 SourceByMetadata(source=p.source, metadata_value=p.value) 

159 for p in db.execute(stmt) 

160 ]