from __future__ import annotations from typing import Any, Dict, List from django import forms from django.db.models import Q from django.http import HttpRequest, HttpResponse from django.shortcuts import get_object_or_404, redirect from django.views import View from django.views.generic.detail import SingleObjectMixin from django.views.generic.edit import FormMixin from django.views.generic.list import MultipleObjectMixin from inertia import render from labs.models import User, UserWrangled from labs.receivers.user_wrangled import get_business_region, get_organization from .base import PermissionRequiredAdminMixin class UserWrangledEditForm(forms.ModelForm): class Meta: model = UserWrangled fields = [ "organization", "pingcaper_role", "business_region", ] class UserWrangledListView(PermissionRequiredAdminMixin, MultipleObjectMixin, View): """List view for UserWrangled data, indexed by user.""" model = User permission_required = "labs.view_userwrangled" paginate_by = 20 page_kwarg = "page" ordering = "-date_joined" sortable_fields = { "id": "id", "email": "email", "organization": "userwrangled__organization", "pingcaper_role": "userwrangled__pingcaper_role", "business_region": "userwrangled__business_region", "updated_at": "userwrangled__updated_at", } search_fields = [ "email", "userwrangled__organization", "userwrangled__pingcaper_role", ] component = "user-wrangled/Index" def get_ordering_from_request(self) -> List[str]: sort_key = self.request.GET.get("sort", "").strip() if not sort_key: if not self.ordering: return [] if isinstance(self.ordering, (list, tuple)): return list(self.ordering) return [self.ordering] sort_field = self.sortable_fields.get(sort_key) if not sort_field: if not self.ordering: return [] if isinstance(self.ordering, (list, tuple)): return list(self.ordering) return [self.ordering] order = self.request.GET.get("order", "").strip() if order not in ("ascend", "descend"): order = "ascend" prefix = "-" if order == "descend" else "" ordering = [f"{prefix}{sort_field}"] # Keep ordering deterministic for pagination when the primary field ties. if sort_field != "id": ordering.append("id") return ordering def get_queryset(self): queryset = self.model._default_manager.all() query = self.request.GET.get("q", "").strip() if query: conditions = Q() for field in self.search_fields: conditions |= Q(**{f"{field}__icontains": query}) queryset = queryset.filter(conditions) ordering = self.get_ordering_from_request() if ordering: queryset = queryset.order_by(*ordering) return queryset def serialize_user(self, user: User, wrangled: UserWrangled | None) -> Dict[str, Any]: return { "id": user.id, "email": user.email, "organization": wrangled.organization if wrangled else None, "pingcaper_role": wrangled.pingcaper_role if wrangled else None, "business_region": wrangled.business_region if wrangled else None, "updated_at": wrangled.updated_at.isoformat() if wrangled and wrangled.updated_at else None, } def get(self, request: HttpRequest) -> HttpResponse: self.object_list = self.get_queryset() paginate_by = self.get_paginate_by(self.object_list) if paginate_by: paginator, page, object_list, is_paginated = self.paginate_queryset( self.object_list, paginate_by ) else: paginator = None page = None object_list = self.object_list is_paginated = False user_ids = [u.id for u in object_list] wrangled_by_user_id = { uw.user_id: uw for uw in UserWrangled.objects.filter(user_id__in=user_ids).all() } users: List[Dict[str, Any]] = [ self.serialize_user(u, wrangled_by_user_id.get(u.id)) for u in object_list ] pagination: Dict[str, Any] | None = None if paginator and page: pagination = { "page": page.number, "per_page": paginator.per_page, "total": paginator.count, "num_pages": paginator.num_pages, "is_paginated": is_paginated, } sort_key = request.GET.get("sort", "").strip() if sort_key not in self.sortable_fields: sort_key = None sort_order = request.GET.get("order", "").strip() if sort_key: if sort_order not in ("ascend", "descend"): sort_order = "ascend" else: sort_order = None return render( request, self.component, { "users": users, "filters": { "email": request.GET.get("email", "").strip(), "q": request.GET.get("q", "").strip(), "sort": sort_key, "order": sort_order, }, "pagination": pagination, }, ) class UserWrangledDetailView( PermissionRequiredAdminMixin, FormMixin, SingleObjectMixin, View ): """Detail/Edit view for a user's UserWrangled record.""" model = UserWrangled permission_required = "labs.view_userwrangled" form_class = UserWrangledEditForm component = "user-wrangled/Detail" def get_object(self, queryset=None) -> UserWrangled: user = get_object_or_404(User, pk=self.kwargs.get("pk")) obj, _ = UserWrangled.objects.get_or_create( user=user, defaults={ "organization": get_organization(user), "business_region": get_business_region(user), }, ) return obj def serialize_user(self, user: User) -> Dict[str, Any]: return { "id": user.id, "email": user.email, } def serialize_user_wrangled(self, wrangled: UserWrangled) -> Dict[str, Any]: return { "organization": wrangled.organization, "pingcaper_role": wrangled.pingcaper_role, "business_region": wrangled.business_region, "created_at": wrangled.created_at.isoformat() if wrangled.created_at else None, "updated_at": wrangled.updated_at.isoformat() if wrangled.updated_at else None, } def get_form_kwargs(self) -> Dict[str, Any]: kwargs = super().get_form_kwargs() kwargs["instance"] = self.object return kwargs def get(self, request: HttpRequest, pk: int) -> HttpResponse: self.object = self.get_object() return render( request, self.component, { "user": self.serialize_user(self.object.user), "wrangled": self.serialize_user_wrangled(self.object), }, ) def post(self, request: HttpRequest, pk: int) -> HttpResponse: if not request.user.has_perm("labs.change_userwrangled"): return redirect(f"/admin/user-wrangled/{pk}/") self.object = self.get_object() form = self.get_form() if form.is_valid(): return self.form_valid(form) return self.form_invalid(form) def form_valid(self, form: forms.ModelForm) -> HttpResponse: form.save() self.request.session["flash_success"] = "User wrangled updated" return redirect(f"/admin/user-wrangled/{self.object.user_id}/") def form_invalid(self, form: forms.ModelForm) -> HttpResponse: return render( self.request, self.component, { "user": self.serialize_user(self.object.user), "wrangled": self.serialize_user_wrangled(self.object), "errors": form.errors, "form": form.data, }, )