Coverage for src/cstlcore/assets/router.py: 73%

1import uuid 

2from io import BytesIO 

3 

4from fastapi import APIRouter, Depends, File, HTTPException, Query 

5from fastapi.responses import StreamingResponse 

6from sqlmodel import Session, select 

7from rapidfuzz import fuzz 

8 

9from cstlcore.assets.models import Asset, AssetCreate, AssetPublic, AssetUpdate 

10from cstlcore.auth.dependencies import get_current_user 

11from cstlcore.constellations.models import Constellation 

12from cstlcore.database.dependencies import get_session 

13from cstlcore.memberships.dependencies import require_read_access, require_write_access 

14from cstlcore.users.models import User 

15import unicodedata 

16 

17def sanitize_filename(filename: str) -> str: 

18 out: list[str] = [] 

19 for ch in filename: 

20 try: 

21 # si le caractère passe en latin-1, on le garde tel quel 

22 ch.encode("latin-1") 

23 out.append(ch) 

24 except UnicodeEncodeError: 

25 # sinon, on tente de "décomposer" le caractère (é -> e + ´) 

26 decomposed = unicodedata.normalize("NFKD", ch) 

27 base = "".join(c for c in decomposed if ord(c) < 128) 

28 if base: 

29 out.append(base) # on garde la base ASCII (é → e, ü → u) 

30 else: 

31 out.append("_") # fallback safe 

32 return "".join(out) 

33 

34router = APIRouter() 

35 

36 

37@router.get( 

38 "/constellations/{constellation_id}/assets", response_model=list[AssetPublic] 

39) 

40async def get_all_assets( 

41 constellation: Constellation = Depends(require_read_access), 

42 session: Session = Depends(get_session), 

43): 

44 db_assets = session.exec( 

45 select(Asset).where(Asset.constellation_id == constellation.id) 

46 ).all() 

47 return db_assets 

48 

49 

50@router.get( 

51 "/constellations/{constellation_id}/assets/search", 

52 response_model=list[AssetPublic], 

53) 

54async def search_assets( 

55 search_query: str = Query(..., description="The search query"), 

56 limit: int = Query(default=100, ge=0, description="The maximum number of items to return"), 

57 page: int = Query(default=1, ge=1, description="The page number to return"), 

58 constellation: Constellation = Depends(require_read_access), 

59 session: Session = Depends(get_session), 

60) -> list[Asset]: 

61 """Search Assets by name within a constellation.""" 

62 # Get all assets from the DB 

63 all_assets = session.exec( 

64 select(Asset).where( 

65 Asset.constellation_id == constellation.id 

66 ) 

67 ).all() 

68 

69 # Then filter search using fuzzy matching (fuzz.partial_ratio >= 80) 

70 matched_assets = [] 

71 for asset in all_assets: 

72 score = fuzz.partial_ratio(search_query.lower(), asset.name.lower()) 

73 if score >= 80: 

74 matched_assets.append((asset, score)) 

75 

76 # Sort by score descending 

77 matched_assets.sort(key=lambda x: x[1], reverse=True) 

78 

79 # Paginate 

80 start_index = (page - 1) * limit 

81 end_index = start_index + limit 

82 if start_index >= len(matched_assets): 

83 return [] 

84 paginated_assets = [asset for asset, score in matched_assets[start_index:end_index]] 

85 

86 return paginated_assets 

87 

88 

89@router.get( 

90 "/constellations/{constellation_id}/assets/{asset_id}", 

91 response_model=AssetPublic, 

92) 

93async def get_asset( 

94 asset_id: uuid.UUID, 

95 constellation: Constellation = Depends(require_read_access), 

96 session: Session = Depends(get_session), 

97): 

98 db_asset = session.exec( 

99 select(Asset).where( 

100 Asset.id == asset_id, Asset.constellation_id == constellation.id 

101 ) 

102 ).first() 

103 if not db_asset: 

104 raise HTTPException(status_code=404, detail="Asset not found") 

105 

106 return db_asset 

107 

108 

109@router.get( 

110 "/constellations/{constellation_id}/assets/{asset_id}/content", 

111 response_class=StreamingResponse, 

112) 

113async def get_asset_content( 

114 asset_id: uuid.UUID, 

115 constellation: Constellation = Depends(require_read_access), 

116 session: Session = Depends(get_session), 

117): 

118 db_asset = session.exec( 

119 select(Asset).where( 

120 Asset.id == asset_id, Asset.constellation_id == constellation.id 

121 ) 

122 ).first() 

123 if not db_asset: 

124 raise HTTPException(status_code=404, detail="Asset not found") 

125 

126 return StreamingResponse( 

127 BytesIO(db_asset.content), 

128 media_type=db_asset.mime_type, 

129 headers={"Content-Disposition": f'attachment; filename="{db_asset.name}"'}, 

130 ) 

131 

132 

133@router.post( 

134 "/constellations/{constellation_id}/assets", 

135 response_model=AssetPublic, 

136 status_code=201, 

137) 

138async def create_asset( 

139 asset: AssetCreate = File(), 

140 user: User = Depends(get_current_user), 

141 constellation: Constellation = Depends(require_write_access), 

142 session: Session = Depends(get_session), 

143): 

144 content = await asset.file.read() 

145 

146 # Sanitize the file name 

147 asset.file.filename = sanitize_filename(asset.file.filename) 

148 

149 db_asset = Asset.model_validate( 

150 asset, 

151 update={ 

152 "constellation_id": constellation.id, 

153 "owner_id": user.id, 

154 "content": content, 

155 "name": sanitize_filename(asset.name) # Sanitize the asset name 

156 }, 

157 ) 

158 session.add(db_asset) 

159 session.commit() 

160 session.refresh(db_asset) 

161 return db_asset 

162 

163 

164@router.put( 

165 "/constellations/{constellation_id}/assets/{asset_id}", 

166 response_model=AssetPublic, 

167) 

168async def update_asset( 

169 asset_id: uuid.UUID, 

170 asset: AssetUpdate, 

171 constellation: Constellation = Depends(require_write_access), 

172 session: Session = Depends(get_session), 

173): 

174 asset_data = asset.model_dump(exclude_unset=True) 

175 db_asset = session.exec( 

176 select(Asset).where( 

177 Asset.id == asset_id, Asset.constellation_id == constellation.id 

178 ) 

179 ).first() 

180 if not db_asset: 

181 raise HTTPException(status_code=404, detail="Asset not found") 

182 

183 db_asset.sqlmodel_update(asset_data) 

184 

185 session.add(db_asset) 

186 session.commit() 

187 session.refresh(db_asset) 

188 return db_asset 

189 

190 

191@router.delete( 

192 "/constellations/{constellation_id}/assets/{asset_id}", 

193) 

194async def delete_asset( 

195 asset_id: uuid.UUID, 

196 constellation: Constellation = Depends(require_write_access), 

197 session: Session = Depends(get_session), 

198): 

199 db_asset = session.exec( 

200 select(Asset).where( 

201 Asset.id == asset_id, Asset.constellation_id == constellation.id 

202 ) 

203 ).first() 

204 if not db_asset: 

205 raise HTTPException(status_code=404, detail="Asset not found") 

206 

207 session.delete(db_asset) 

208 session.commit() 

209 return {"ok": True}