Coverage for apps / recipes / api_user.py: 94%

1""" 

2User feature API endpoints: favorites, collections, history. 

3 

4All endpoints are scoped to the current profile (from session). 

5""" 

6 

7from typing import List, Optional 

8 

9from django.db import IntegrityError 

10from django.db.models import Max 

11from django.shortcuts import get_object_or_404 

12from ninja import Router, Schema, Status 

13 

14from django_ratelimit.decorators import ratelimit 

15 

16from apps.core.auth import SessionAuth 

17from apps.profiles.utils import get_current_profile 

18 

19from .api import RecipeListOut 

20from .models import ( 

21 Recipe, 

22 RecipeCollection, 

23 RecipeCollectionItem, 

24 RecipeFavorite, 

25 RecipeViewHistory, 

26) 

27 

28# ============================================================================= 

29# Favorites Router 

30# ============================================================================= 

31 

32favorites_router = Router(tags=["favorites"]) 

33 

34 

35class FavoriteIn(Schema): 

36 recipe_id: int 

37 

38 

39class FavoriteOut(Schema): 

40 recipe: RecipeListOut 

41 created_at: str 

42 

43 @staticmethod 

44 def resolve_created_at(obj): 

45 return obj.created_at.isoformat() 

46 

47 

48class ErrorOut(Schema): 

49 detail: str 

50 

51 

52@favorites_router.get("/", response=List[FavoriteOut], auth=SessionAuth()) 

53def list_favorites(request): 

54 """List all favorites for the current profile.""" 

55 profile = get_current_profile(request) 

56 return RecipeFavorite.objects.filter(profile=profile).select_related("recipe") 

57 

58 

59@favorites_router.post( 

60 "/", response={201: FavoriteOut, 400: ErrorOut, 404: ErrorOut, 429: ErrorOut}, auth=SessionAuth() 

61) 

62@ratelimit(key="ip", rate="60/h", method="POST", block=False) 

63def add_favorite(request, payload: FavoriteIn): 

64 """Add a recipe to favorites.""" 

65 if getattr(request, "limited", False): 

66 return Status(429, {"detail": "Too many requests. Please try again later."}) 

67 profile = get_current_profile(request) 

68 recipe = get_object_or_404(Recipe, id=payload.recipe_id) 

69 

70 try: 

71 favorite = RecipeFavorite.objects.create(profile=profile, recipe=recipe) 

72 return Status(201, favorite) 

73 except IntegrityError: 

74 return Status(400, {"detail": "Recipe is already a favorite"}) 

75 

76 

77@favorites_router.delete("/{recipe_id}/", response={204: None, 404: ErrorOut}, auth=SessionAuth()) 

78def remove_favorite(request, recipe_id: int): 

79 """Remove a recipe from favorites.""" 

80 profile = get_current_profile(request) 

81 favorite = get_object_or_404(RecipeFavorite, profile=profile, recipe_id=recipe_id) 

82 favorite.delete() 

83 return Status(204, None) 

84 

85 

86# ============================================================================= 

87# Collections Router 

88# ============================================================================= 

89 

90collections_router = Router(tags=["collections"]) 

91 

92 

93class CollectionIn(Schema): 

94 name: str 

95 description: str = "" 

96 

97 

98class CollectionItemIn(Schema): 

99 recipe_id: int 

100 

101 

102class CollectionItemOut(Schema): 

103 recipe: RecipeListOut 

104 order: int 

105 added_at: str 

106 

107 @staticmethod 

108 def resolve_added_at(obj): 

109 return obj.added_at.isoformat() 

110 

111 

112class CollectionOut(Schema): 

113 id: int 

114 name: str 

115 description: str 

116 recipe_count: int 

117 cover_image: Optional[str] 

118 created_at: str 

119 updated_at: str 

120 

121 @staticmethod 

122 def resolve_recipe_count(obj): 

123 return obj.items.count() 

124 

125 @staticmethod 

126 def resolve_cover_image(obj): 

127 first_item = obj.items.select_related("recipe").first() 

128 if first_item: 

129 recipe = first_item.recipe 

130 if recipe.image: 

131 return recipe.image.url 

132 if recipe.image_url: 

133 return recipe.image_url 

134 return None 

135 

136 @staticmethod 

137 def resolve_created_at(obj): 

138 return obj.created_at.isoformat() 

139 

140 @staticmethod 

141 def resolve_updated_at(obj): 

142 return obj.updated_at.isoformat() 

143 

144 

145class CollectionDetailOut(Schema): 

146 id: int 

147 name: str 

148 description: str 

149 recipes: List[CollectionItemOut] 

150 created_at: str 

151 updated_at: str 

152 

153 @staticmethod 

154 def resolve_recipes(obj): 

155 return obj.items.select_related("recipe").all() 

156 

157 @staticmethod 

158 def resolve_created_at(obj): 

159 return obj.created_at.isoformat() 

160 

161 @staticmethod 

162 def resolve_updated_at(obj): 

163 return obj.updated_at.isoformat() 

164 

165 

166@collections_router.get("/", response=List[CollectionOut], auth=SessionAuth()) 

167def list_collections(request): 

168 """List all collections for the current profile.""" 

169 profile = get_current_profile(request) 

170 return RecipeCollection.objects.filter(profile=profile).prefetch_related("items") 

171 

172 

173@collections_router.post("/", response={201: CollectionOut, 400: ErrorOut, 429: ErrorOut}, auth=SessionAuth()) 

174@ratelimit(key="ip", rate="60/h", method="POST", block=False) 

175def create_collection(request, payload: CollectionIn): 

176 """Create a new collection.""" 

177 if getattr(request, "limited", False): 

178 return Status(429, {"detail": "Too many requests. Please try again later."}) 

179 profile = get_current_profile(request) 

180 

181 try: 

182 collection = RecipeCollection.objects.create( 

183 profile=profile, 

184 name=payload.name, 

185 description=payload.description, 

186 ) 

187 return Status(201, collection) 

188 except IntegrityError: 

189 return Status(400, {"detail": "A collection with this name already exists"}) 

190 

191 

192@collections_router.get("/{collection_id}/", response={200: CollectionDetailOut, 404: ErrorOut}, auth=SessionAuth()) 

193def get_collection(request, collection_id: int): 

194 """Get a collection with its recipes.""" 

195 profile = get_current_profile(request) 

196 collection = get_object_or_404(RecipeCollection, id=collection_id, profile=profile) 

197 return collection 

198 

199 

200@collections_router.put( 

201 "/{collection_id}/", response={200: CollectionOut, 400: ErrorOut, 404: ErrorOut}, auth=SessionAuth() 

202) 

203def update_collection(request, collection_id: int, payload: CollectionIn): 

204 """Update a collection.""" 

205 profile = get_current_profile(request) 

206 collection = get_object_or_404(RecipeCollection, id=collection_id, profile=profile) 

207 

208 try: 

209 collection.name = payload.name 

210 collection.description = payload.description 

211 collection.save() 

212 return collection 

213 except IntegrityError: 

214 return Status(400, {"detail": "A collection with this name already exists"}) 

215 

216 

217@collections_router.delete("/{collection_id}/", response={204: None, 404: ErrorOut}, auth=SessionAuth()) 

218def delete_collection(request, collection_id: int): 

219 """Delete a collection.""" 

220 profile = get_current_profile(request) 

221 collection = get_object_or_404(RecipeCollection, id=collection_id, profile=profile) 

222 collection.delete() 

223 return Status(204, None) 

224 

225 

226@collections_router.post( 

227 "/{collection_id}/recipes/", response={201: CollectionItemOut, 400: ErrorOut, 404: ErrorOut}, auth=SessionAuth() 

228) 

229def add_recipe_to_collection(request, collection_id: int, payload: CollectionItemIn): 

230 """Add a recipe to a collection.""" 

231 profile = get_current_profile(request) 

232 collection = get_object_or_404(RecipeCollection, id=collection_id, profile=profile) 

233 recipe = get_object_or_404(Recipe, id=payload.recipe_id) 

234 

235 # Get the next order value 

236 max_order = collection.items.aggregate(max_order=Max("order"))["max_order"] 

237 next_order = (max_order or 0) + 1 

238 

239 try: 

240 item = RecipeCollectionItem.objects.create( 

241 collection=collection, 

242 recipe=recipe, 

243 order=next_order, 

244 ) 

245 return Status(201, item) 

246 except IntegrityError: 

247 return Status(400, {"detail": "Recipe is already in this collection"}) 

248 

249 

250@collections_router.delete( 

251 "/{collection_id}/recipes/{recipe_id}/", response={204: None, 404: ErrorOut}, auth=SessionAuth() 

252) 

253def remove_recipe_from_collection(request, collection_id: int, recipe_id: int): 

254 """Remove a recipe from a collection.""" 

255 profile = get_current_profile(request) 

256 collection = get_object_or_404(RecipeCollection, id=collection_id, profile=profile) 

257 item = get_object_or_404(RecipeCollectionItem, collection=collection, recipe_id=recipe_id) 

258 item.delete() 

259 return Status(204, None) 

260 

261 

262# ============================================================================= 

263# History Router 

264# ============================================================================= 

265 

266history_router = Router(tags=["history"]) 

267 

268 

269class HistoryIn(Schema): 

270 recipe_id: int 

271 

272 

273class HistoryOut(Schema): 

274 recipe: RecipeListOut 

275 viewed_at: str 

276 

277 @staticmethod 

278 def resolve_viewed_at(obj): 

279 return obj.viewed_at.isoformat() 

280 

281 

282@history_router.get("/", response=List[HistoryOut], auth=SessionAuth()) 

283def list_history(request, limit: int = 6): 

284 """ 

285 Get recently viewed recipes for the current profile. 

286 

287 Returns up to 6 most recent recipes by default (max 100). 

288 """ 

289 profile = get_current_profile(request) 

290 limit = min(max(limit, 1), 100) 

291 return RecipeViewHistory.objects.filter(profile=profile).select_related("recipe")[:limit] 

292 

293 

294@history_router.post("/", response={200: HistoryOut, 201: HistoryOut, 404: ErrorOut, 429: ErrorOut}, auth=SessionAuth()) 

295@ratelimit(key="ip", rate="120/h", method="POST", block=False) 

296def record_view(request, payload: HistoryIn): 

297 """ 

298 Record a recipe view. 

299 

300 If the recipe was already viewed, updates the timestamp. 

301 """ 

302 if getattr(request, "limited", False): 

303 return Status(429, {"detail": "Too many requests. Please try again later."}) 

304 profile = get_current_profile(request) 

305 recipe = get_object_or_404(Recipe, id=payload.recipe_id) 

306 

307 history, created = RecipeViewHistory.objects.update_or_create( 

308 profile=profile, 

309 recipe=recipe, 

310 defaults={}, # viewed_at auto-updates due to auto_now 

311 ) 

312 

313 status_code = 201 if created else 200 

314 return Status(status_code, history) 

315 

316 

317@history_router.delete("/", response={204: None}, auth=SessionAuth()) 

318def clear_history(request): 

319 """Clear all view history for the current profile.""" 

320 profile = get_current_profile(request) 

321 RecipeViewHistory.objects.filter(profile=profile).delete() 

322 return Status(204, None)