Coverage for apps / core / api.py: 91%

1"""System API for administrative operations like database reset.""" 

2 

3import os 

4import shutil 

5 

6from django.conf import settings 

7from django.core.cache import cache 

8from django.core.management import call_command 

9from django.contrib.sessions.models import Session 

10from ninja import Router, Schema 

11 

12from apps.profiles.models import Profile 

13from apps.recipes.models import ( 

14 Recipe, 

15 RecipeFavorite, 

16 RecipeCollection, 

17 RecipeCollectionItem, 

18 RecipeViewHistory, 

19 SearchSource, 

20 ServingAdjustment, 

21 CachedSearchImage, 

22) 

23from apps.ai.models import AIDiscoverySuggestion, AIPrompt 

24 

25router = Router(tags=["system"]) 

26 

27 

28class HealthSchema(Schema): 

29 status: str 

30 database: str 

31 

32 

33@router.get("/health/", response=HealthSchema) 

34def health_check(request): 

35 """Simple health check for container orchestration.""" 

36 from django.db import connection 

37 

38 try: 

39 with connection.cursor() as cursor: 

40 cursor.execute("SELECT 1") 

41 return {"status": "healthy", "database": "ok"} 

42 except Exception: 

43 return {"status": "unhealthy", "database": "error"} 

44 

45 

46class DataCountsSchema(Schema): 

47 profiles: int 

48 recipes: int 

49 recipe_images: int 

50 favorites: int 

51 collections: int 

52 collection_items: int 

53 view_history: int 

54 ai_suggestions: int 

55 serving_adjustments: int 

56 cached_search_images: int 

57 

58 

59class ResetPreviewSchema(Schema): 

60 data_counts: DataCountsSchema 

61 preserved: list[str] 

62 warnings: list[str] 

63 

64 

65class ResetConfirmSchema(Schema): 

66 confirmation_text: str # Must be "RESET" 

67 

68 

69class ErrorSchema(Schema): 

70 error: str 

71 message: str 

72 

73 

74class ResetSuccessSchema(Schema): 

75 success: bool 

76 message: str 

77 actions_performed: list[str] 

78 

79 

80@router.get("/reset-preview/", response=ResetPreviewSchema) 

81def get_reset_preview(request): 

82 """Get summary of data that will be deleted on reset.""" 

83 return { 

84 "data_counts": { 

85 "profiles": Profile.objects.count(), 

86 "recipes": Recipe.objects.count(), 

87 "recipe_images": Recipe.objects.exclude(image="").exclude(image__isnull=True).count(), 

88 "favorites": RecipeFavorite.objects.count(), 

89 "collections": RecipeCollection.objects.count(), 

90 "collection_items": RecipeCollectionItem.objects.count(), 

91 "view_history": RecipeViewHistory.objects.count(), 

92 "ai_suggestions": AIDiscoverySuggestion.objects.count(), 

93 "serving_adjustments": ServingAdjustment.objects.count(), 

94 "cached_search_images": CachedSearchImage.objects.count(), 

95 }, 

96 "preserved": [ 

97 "Search source configurations", 

98 "AI prompt templates", 

99 "Application settings", 

100 ], 

101 "warnings": [ 

102 "All user data will be permanently deleted", 

103 "All recipe images will be removed from storage", 

104 "This action cannot be undone", 

105 ], 

106 } 

107 

108 

109@router.post("/reset/", response={200: ResetSuccessSchema, 400: ErrorSchema}) 

110def reset_database(request, data: ResetConfirmSchema): 

111 """ 

112 Completely reset the database to factory state. 

113 

114 Requires confirmation_text="RESET" to proceed. 

115 """ 

116 if data.confirmation_text != "RESET": 

117 return 400, { 

118 "error": "invalid_confirmation", 

119 "message": "Type RESET to confirm", 

120 } 

121 

122 try: 

123 # 1. Clear database tables (order matters for FK constraints) 

124 # Start with leaf tables that depend on others 

125 AIDiscoverySuggestion.objects.all().delete() 

126 ServingAdjustment.objects.all().delete() 

127 RecipeViewHistory.objects.all().delete() 

128 RecipeCollectionItem.objects.all().delete() 

129 RecipeCollection.objects.all().delete() 

130 RecipeFavorite.objects.all().delete() 

131 CachedSearchImage.objects.all().delete() 

132 

133 # Delete all recipes (this will cascade to related items) 

134 Recipe.objects.all().delete() 

135 

136 # Delete all profiles 

137 Profile.objects.all().delete() 

138 

139 # Reset SearchSource failure counters (keep selectors) 

140 SearchSource.objects.all().update( 

141 consecutive_failures=0, 

142 needs_attention=False, 

143 last_validated_at=None, 

144 ) 

145 

146 # 2. Clear recipe images 

147 images_dir = os.path.join(settings.MEDIA_ROOT, "recipe_images") 

148 if os.path.exists(images_dir): 

149 shutil.rmtree(images_dir) 

150 os.makedirs(images_dir) # Recreate empty directory 

151 

152 # Clear cached search images 

153 search_images_dir = os.path.join(settings.MEDIA_ROOT, "search_images") 

154 if os.path.exists(search_images_dir): 

155 shutil.rmtree(search_images_dir) 

156 os.makedirs(search_images_dir) # Recreate empty directory 

157 

158 # 3. Clear Django cache 

159 cache.clear() 

160 

161 # 4. Clear all sessions 

162 Session.objects.all().delete() 

163 

164 # 5. Re-run migrations (ensures clean state) 

165 call_command("migrate", verbosity=0) 

166 

167 # 6. Re-seed default data 

168 try: 

169 call_command("seed_search_sources", verbosity=0) 

170 except Exception: 

171 pass # Command may not exist yet 

172 

173 try: 

174 call_command("seed_ai_prompts", verbosity=0) 

175 except Exception: 

176 pass # Command may not exist yet 

177 

178 return { 

179 "success": True, 

180 "message": "Database reset complete", 

181 "actions_performed": [ 

182 "Deleted all user profiles", 

183 "Deleted all recipes and images", 

184 "Cleared all favorites and collections", 

185 "Cleared all view history", 

186 "Cleared all AI cache data", 

187 "Cleared all cached search images", 

188 "Reset search source counters", 

189 "Cleared application cache", 

190 "Cleared all sessions", 

191 "Re-ran database migrations", 

192 "Restored default seed data", 

193 ], 

194 } 

195 

196 except Exception as e: 

197 return 400, {"error": "reset_failed", "message": str(e)}