Coverage for apps / profiles / api.py: 85%
179 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-12 10:49 +0000
1import os
2from datetime import datetime
3from typing import List, Optional
5from django.conf import settings
6from django.db.models import Count, Q
7from django_ratelimit.decorators import ratelimit
8from ninja import Router, Schema, Status
10from apps.core.auth import AdminAuth, SessionAuth
11from .models import Profile
13router = Router(tags=["profiles"])
16class ProfileIn(Schema):
17 name: str
18 avatar_color: str = ""
19 theme: str = "light"
20 unit_preference: str = "metric"
23class ProfileOut(Schema):
24 id: int
25 name: str
26 avatar_color: str
27 theme: str
28 unit_preference: str
29 is_admin: Optional[bool] = None
32class ProfileStatsSchema(Schema):
33 favorites: int
34 collections: int
35 collection_items: int
36 remixes: int
37 view_history: int
38 scaling_cache: int
39 discover_cache: int
42class ProfileWithStatsSchema(Schema):
43 id: int
44 name: str
45 avatar_color: str
46 theme: str
47 unit_preference: str
48 unlimited_ai: bool = False
49 created_at: datetime
50 stats: ProfileStatsSchema
53class DeletionDataSchema(Schema):
54 remixes: int
55 remix_images: int
56 favorites: int
57 collections: int
58 collection_items: int
59 view_history: int
60 scaling_cache: int
61 discover_cache: int
64class ProfileSummarySchema(Schema):
65 id: int
66 name: str
67 avatar_color: str
68 created_at: datetime
71class DeletionPreviewSchema(Schema):
72 profile: ProfileSummarySchema
73 data_to_delete: DeletionDataSchema
74 warnings: List[str]
77class SetUnlimitedIn(Schema):
78 unlimited: bool
81class RenameIn(Schema):
82 name: str
85class ErrorSchema(Schema):
86 error: str
87 message: str
90def _resolve_authenticated_user(request):
91 """Resolve the authenticated user in passkey mode. Returns (user, profile) or (None, None)."""
92 user = getattr(request, "user", None)
93 if user and getattr(user, "is_authenticated", False):
94 try:
95 return user, user.profile
96 except Profile.DoesNotExist:
97 return None, None
99 # Fallback: resolve from session profile_id
100 profile_id = request.session.get("profile_id")
101 if profile_id:
102 try:
103 p = Profile.objects.select_related("user").get(id=profile_id)
104 if p.user and p.user.is_active:
105 return p.user, p
106 except Profile.DoesNotExist:
107 pass
108 return None, None
111def _check_profile_ownership(request, profile_id):
112 """In passkey mode, verify the user owns the profile (or is admin). Returns error tuple or None."""
113 if settings.AUTH_MODE != "passkey":
114 return None
115 user, own_profile = _resolve_authenticated_user(request)
116 if not user:
117 return Status(404, {"error": "not_found", "message": "Profile not found"})
118 if user.is_staff:
119 return None # Admin can access any profile
120 if not own_profile or own_profile.id != profile_id:
121 return Status(404, {"error": "not_found", "message": "Profile not found"})
122 return None
125@router.get(
126 "/",
127 response=List[ProfileWithStatsSchema],
128 auth=[SessionAuth()] if settings.AUTH_MODE == "passkey" else None,
129)
130def list_profiles(request):
131 """List all profiles with stats.
133 Passkey mode: returns only current user's profile (admin sees all).
134 Home mode: returns all profiles (no auth required — this is the profile selection screen).
135 """
136 from apps.recipes.models import RecipeCollectionItem
138 profiles = Profile.objects.annotate(
139 favorites_count=Count("favorites", distinct=True),
140 collections_count=Count("collections", distinct=True),
141 remixes_count=Count("remixes", filter=Q(remixes__is_remix=True), distinct=True),
142 view_history_count=Count("view_history", distinct=True),
143 scaling_cache_count=Count("serving_adjustments", distinct=True),
144 discover_cache_count=Count("ai_discovery_suggestions", distinct=True),
145 ).order_by("-created_at")
147 if settings.AUTH_MODE == "passkey":
148 user, _ = _resolve_authenticated_user(request)
149 if not user:
150 return []
151 if not user.is_staff:
152 profiles = profiles.filter(user=user)
154 result = []
155 for p in profiles:
156 collection_items_count = RecipeCollectionItem.objects.filter(collection__profile=p).count()
158 result.append(
159 ProfileWithStatsSchema(
160 id=p.id,
161 name=p.name,
162 avatar_color=p.avatar_color,
163 theme=p.theme,
164 unit_preference=p.unit_preference,
165 unlimited_ai=p.unlimited_ai,
166 created_at=p.created_at,
167 stats=ProfileStatsSchema(
168 favorites=p.favorites_count,
169 collections=p.collections_count,
170 collection_items=collection_items_count,
171 remixes=p.remixes_count,
172 view_history=p.view_history_count,
173 scaling_cache=p.scaling_cache_count,
174 discover_cache=p.discover_cache_count,
175 ),
176 )
177 )
178 return result
181@router.post("/", response={201: ProfileOut, 404: ErrorSchema, 429: ErrorSchema})
182@ratelimit(key="ip", rate="10/h", method="POST", block=False)
183def create_profile(request, payload: ProfileIn):
184 """Create a new profile. Only available in home mode."""
185 if getattr(request, "limited", False):
186 return Status(429, {"error": "rate_limited", "message": "Too many requests. Please try again later."})
187 if settings.AUTH_MODE != "home":
188 return Status(404, {"error": "not_found", "message": "Not found"})
189 data = payload.dict()
190 if not data.get("avatar_color"):
191 data["avatar_color"] = Profile.next_avatar_color()
192 profile = Profile.objects.create(**data)
193 return Status(201, profile)
196@router.get("/{profile_id}/", response={200: ProfileOut, 404: ErrorSchema}, auth=SessionAuth())
197def get_profile(request, profile_id: int):
198 """Get a profile by ID. Public mode: own profile only (admin: any)."""
199 ownership_error = _check_profile_ownership(request, profile_id)
200 if ownership_error:
201 return ownership_error
202 try:
203 return Profile.objects.get(id=profile_id)
204 except Profile.DoesNotExist:
205 return Status(404, {"error": "not_found", "message": "Profile not found"})
208@router.put("/{profile_id}/", response={200: ProfileOut, 404: ErrorSchema}, auth=SessionAuth())
209def update_profile(request, profile_id: int, payload: ProfileIn):
210 """Update a profile. Public mode: own profile only (admin: any)."""
211 ownership_error = _check_profile_ownership(request, profile_id)
212 if ownership_error:
213 return ownership_error
214 try:
215 profile = Profile.objects.get(id=profile_id)
216 except Profile.DoesNotExist:
217 return Status(404, {"error": "not_found", "message": "Profile not found"})
218 for key, value in payload.dict().items():
219 setattr(profile, key, value)
220 profile.save()
221 return profile
224@router.get(
225 "/{profile_id}/deletion-preview/", response={200: DeletionPreviewSchema, 404: ErrorSchema}, auth=SessionAuth()
226)
227def get_deletion_preview(request, profile_id: int):
228 """Get summary of data that will be deleted. Public mode: own profile only."""
229 from apps.ai.models import AIDiscoverySuggestion
230 from apps.recipes.models import (
231 Recipe,
232 RecipeCollection,
233 RecipeCollectionItem,
234 RecipeFavorite,
235 RecipeViewHistory,
236 ServingAdjustment,
237 )
239 ownership_error = _check_profile_ownership(request, profile_id)
240 if ownership_error:
241 return ownership_error
243 try:
244 profile = Profile.objects.get(id=profile_id)
245 except Profile.DoesNotExist:
246 return Status(404, {"error": "not_found", "message": "Profile not found"})
248 remixes = Recipe.objects.filter(is_remix=True, remix_profile=profile)
249 favorites = RecipeFavorite.objects.filter(profile=profile)
250 collections = RecipeCollection.objects.filter(profile=profile)
251 collection_items = RecipeCollectionItem.objects.filter(collection__profile=profile)
252 view_history = RecipeViewHistory.objects.filter(profile=profile)
253 scaling_cache = ServingAdjustment.objects.filter(profile=profile)
254 discover_cache = AIDiscoverySuggestion.objects.filter(profile=profile)
255 remix_images_count = remixes.exclude(image="").exclude(image__isnull=True).count()
257 return {
258 "profile": {
259 "id": profile.id,
260 "name": profile.name,
261 "avatar_color": profile.avatar_color,
262 "created_at": profile.created_at,
263 },
264 "data_to_delete": {
265 "remixes": remixes.count(),
266 "remix_images": remix_images_count,
267 "favorites": favorites.count(),
268 "collections": collections.count(),
269 "collection_items": collection_items.count(),
270 "view_history": view_history.count(),
271 "scaling_cache": scaling_cache.count(),
272 "discover_cache": discover_cache.count(),
273 },
274 "warnings": [
275 "All remixed recipes will be permanently deleted",
276 "Recipe images for remixes will be removed from storage",
277 "This action cannot be undone",
278 ],
279 }
282@router.delete("/{profile_id}/", response={204: None, 400: ErrorSchema, 404: ErrorSchema}, auth=SessionAuth())
283def delete_profile(request, profile_id: int):
284 """Delete a profile and ALL associated data.
286 In passkey mode: own profile only, also cascades to delete the Django User.
287 """
288 from apps.recipes.models import Recipe
290 ownership_error = _check_profile_ownership(request, profile_id)
291 if ownership_error:
292 return ownership_error
294 try:
295 profile = Profile.objects.get(id=profile_id)
296 except Profile.DoesNotExist:
297 return Status(404, {"error": "not_found", "message": "Profile not found"})
299 current_profile_id = request.session.get("profile_id")
300 if current_profile_id == profile_id:
301 request.session.pop("profile_id", None)
303 # Collect image paths BEFORE cascade delete
304 remix_images = list(
305 Recipe.objects.filter(is_remix=True, remix_profile=profile, image__isnull=False)
306 .exclude(image="")
307 .values_list("image", flat=True)
308 )
310 if settings.AUTH_MODE == "passkey" and profile.user:
311 profile.user.delete()
312 request.session.flush()
313 else:
314 profile.delete()
316 for image_path in remix_images:
317 full_path = os.path.join(settings.MEDIA_ROOT, str(image_path))
318 try:
319 if os.path.exists(full_path):
320 os.remove(full_path)
321 except OSError:
322 pass
324 return Status(204, None)
327@router.post("/{profile_id}/select/", response={200: ProfileOut, 404: dict})
328def select_profile(request, profile_id: int):
329 """Set a profile as the current profile. Only available in home mode."""
330 if settings.AUTH_MODE != "home":
331 return Status(404, {"detail": "Not found"})
332 try:
333 profile = Profile.objects.get(id=profile_id)
334 except Profile.DoesNotExist:
335 request.session.pop("profile_id", None)
336 return Status(404, {"detail": "Profile not found"})
337 request.session["profile_id"] = profile.id
338 return profile
341@router.post("/{profile_id}/set-unlimited/", response={200: dict, 404: ErrorSchema}, auth=AdminAuth())
342def set_unlimited(request, profile_id: int, data: SetUnlimitedIn):
343 """Set or revoke unlimited AI access for a profile. Admin only."""
344 try:
345 profile = Profile.objects.get(id=profile_id)
346 except Profile.DoesNotExist:
347 return Status(404, {"error": "not_found", "message": "Profile not found"})
348 profile.unlimited_ai = data.unlimited
349 profile.save(update_fields=["unlimited_ai"])
350 return {"id": profile.id, "name": profile.name, "unlimited_ai": profile.unlimited_ai}
353@router.patch("/{profile_id}/rename/", response={200: dict, 400: ErrorSchema, 404: ErrorSchema}, auth=AdminAuth())
354def rename_profile(request, profile_id: int, data: RenameIn):
355 """Rename a profile. Admin only."""
356 name = data.name.strip()
357 if not name or len(name) > 100:
358 return Status(400, {"error": "validation_error", "message": "Name must be between 1 and 100 characters"})
359 try:
360 profile = Profile.objects.get(id=profile_id)
361 except Profile.DoesNotExist:
362 return Status(404, {"error": "not_found", "message": "Profile not found"})
363 profile.name = name
364 profile.save(update_fields=["name"])
365 return {"id": profile.id, "name": profile.name, "avatar_color": profile.avatar_color}