|
| 1 | +"""``GET /api/v1/<app>/<model>/<pk>/history/`` — object history. |
| 2 | +
|
| 3 | +Wire contract: ``docs/api-contract.md`` §4 (history sub-resource). |
| 4 | +
|
| 5 | +Surfaces the ``django.contrib.admin.models.LogEntry`` timeline for a |
| 6 | +single object — the same data the legacy admin's *History* button |
| 7 | +shows. Parity (#155): a Django dev's audit trail must be reachable |
| 8 | +from the SPA, and the entries the SPA itself writes (via the create / |
| 9 | +update / delete endpoints, which call ``ModelAdmin.log_*``) show up |
| 10 | +here alongside any earlier HTML-admin entries. |
| 11 | +
|
| 12 | +Hard rules (`SECURITY.md` §3): |
| 13 | +
|
| 14 | +- Rule 1: Staff + ``AdminSite.has_permission`` gate. |
| 15 | +- Rule 3: Model resolved through ``admin.site._registry`` (B-7). |
| 16 | +- Rule 5: Per-object ``has_view_permission`` gate — you can only read |
| 17 | + the history of an object you can view. |
| 18 | +- Rule 10: Object loaded through ``ModelAdmin.get_queryset(request)`` |
| 19 | + — never ``Model.objects.all()`` (B-2). |
| 20 | +- CSRF: GET is safe; no state change. |
| 21 | +""" |
| 22 | + |
| 23 | +from __future__ import annotations |
| 24 | + |
| 25 | +from typing import Any |
| 26 | + |
| 27 | +from django.contrib.admin.models import ADDITION |
| 28 | +from django.contrib.admin.models import CHANGE |
| 29 | +from django.contrib.admin.models import DELETION |
| 30 | +from django.contrib.admin.models import LogEntry |
| 31 | +from django.core.paginator import Paginator |
| 32 | +from django.http import HttpRequest |
| 33 | +from django.http import HttpResponse |
| 34 | +from django.http import JsonResponse |
| 35 | +from django.views.generic import View |
| 36 | + |
| 37 | +from django_admin_react.api.permissions import forbidden_response |
| 38 | +from django_admin_react.api.permissions import is_admin_user |
| 39 | +from django_admin_react.api.registry import get_admin_site |
| 40 | +from django_admin_react.api.registry import resolve_model |
| 41 | +from django_admin_react.api.serializers import label_for |
| 42 | +from django_admin_react.api.writes import load_object_or_none |
| 43 | +from django_admin_react.api.writes import not_found_response |
| 44 | +from django_admin_react.audit import object_log_entries |
| 45 | + |
| 46 | +_ACTION_LABELS = {ADDITION: "addition", CHANGE: "change", DELETION: "deletion"} |
| 47 | + |
| 48 | +_DEFAULT_PAGE_SIZE = 25 |
| 49 | +_MAX_PAGE_SIZE = 200 |
| 50 | + |
| 51 | + |
| 52 | +class HistoryView(View): |
| 53 | + """``GET /api/v1/<app_label>/<model_name>/<pk>/history/``.""" |
| 54 | + |
| 55 | + http_method_names = ["get"] |
| 56 | + |
| 57 | + def get( |
| 58 | + self, |
| 59 | + request: HttpRequest, |
| 60 | + app_label: str, |
| 61 | + model_name: str, |
| 62 | + pk: str, |
| 63 | + *args: Any, |
| 64 | + **kwargs: Any, |
| 65 | + ) -> HttpResponse: |
| 66 | + """Return the paginated ``LogEntry`` timeline for one object. |
| 67 | +
|
| 68 | + Gates: ``is_admin_user`` → ``resolve_model`` → object loaded |
| 69 | + through ``get_queryset`` → ``has_view_permission(obj)``. A |
| 70 | + missing object or unviewable object both return the canonical |
| 71 | + 404 (no oracle distinguishing "doesn't exist" from "you can't |
| 72 | + see it" — ``SECURITY.md`` §3 rule 12). |
| 73 | + """ |
| 74 | + admin_site = get_admin_site() |
| 75 | + if not is_admin_user(request, admin_site=admin_site): |
| 76 | + return forbidden_response(request) |
| 77 | + |
| 78 | + resolved = resolve_model(admin_site, request, app_label, model_name) |
| 79 | + if resolved is None: |
| 80 | + return not_found_response() |
| 81 | + model, model_admin = resolved |
| 82 | + |
| 83 | + obj = load_object_or_none(model, model_admin, request, pk) |
| 84 | + if obj is None: |
| 85 | + return not_found_response() |
| 86 | + |
| 87 | + if not model_admin.has_view_permission(request, obj): |
| 88 | + return forbidden_response(request) |
| 89 | + |
| 90 | + entries = object_log_entries(obj) |
| 91 | + |
| 92 | + paginator = Paginator(entries, _page_size(request)) |
| 93 | + page_number = _page_number(request) |
| 94 | + page = paginator.get_page(page_number) |
| 95 | + |
| 96 | + body = { |
| 97 | + "object": {"pk": obj.pk, "label": label_for(obj)}, |
| 98 | + "entries": [_serialize_entry(e) for e in page.object_list], |
| 99 | + "page": page.number, |
| 100 | + "page_size": paginator.per_page, |
| 101 | + "total": paginator.count, |
| 102 | + "num_pages": paginator.num_pages, |
| 103 | + } |
| 104 | + response = JsonResponse(body, status=200) |
| 105 | + response["Cache-Control"] = "no-store" |
| 106 | + return response |
| 107 | + |
| 108 | + |
| 109 | +def _serialize_entry(entry: LogEntry) -> dict[str, Any]: |
| 110 | + """One ``LogEntry`` → wire shape. |
| 111 | +
|
| 112 | + ``change_message_human`` is Django's own rendered summary |
| 113 | + (``get_change_message``); ``change_message_structured`` is the raw |
| 114 | + JSON list so a SPA can render field-level detail without re-parsing |
| 115 | + the prose. |
| 116 | + """ |
| 117 | + user = entry.user |
| 118 | + return { |
| 119 | + "id": entry.id, |
| 120 | + "action": _ACTION_LABELS.get(entry.action_flag, "unknown"), |
| 121 | + "action_time": entry.action_time.isoformat(), |
| 122 | + "user": None if user is None else {"id": user.pk, "label": str(user)}, |
| 123 | + "change_message_human": entry.get_change_message(), |
| 124 | + "change_message_structured": _structured_message(entry), |
| 125 | + } |
| 126 | + |
| 127 | + |
| 128 | +def _structured_message(entry: LogEntry) -> Any: |
| 129 | + """Return the raw structured change message, or ``[]`` if absent. |
| 130 | +
|
| 131 | + ``LogEntry.change_message`` is a JSON string for entries written by |
| 132 | + modern admin; older / hand-written entries may store free text. |
| 133 | + ``get_change_message`` already handles the prose rendering, so here |
| 134 | + we only surface the structured form when it parses as a list. |
| 135 | + """ |
| 136 | + import json |
| 137 | + |
| 138 | + raw = entry.change_message or "" |
| 139 | + try: |
| 140 | + parsed = json.loads(raw) |
| 141 | + except (ValueError, TypeError): |
| 142 | + return [] |
| 143 | + return parsed if isinstance(parsed, list) else [] |
| 144 | + |
| 145 | + |
| 146 | +def _page_size(request: HttpRequest) -> int: |
| 147 | + """Clamp the ``page_size`` query param to ``[1, _MAX_PAGE_SIZE]``.""" |
| 148 | + raw = request.GET.get("page_size") |
| 149 | + if raw is None: |
| 150 | + return _DEFAULT_PAGE_SIZE |
| 151 | + try: |
| 152 | + value = int(raw) |
| 153 | + except (TypeError, ValueError): |
| 154 | + return _DEFAULT_PAGE_SIZE |
| 155 | + return max(1, min(value, _MAX_PAGE_SIZE)) |
| 156 | + |
| 157 | + |
| 158 | +def _page_number(request: HttpRequest) -> int: |
| 159 | + """Read the ``page`` query param; default 1 on absent / bogus.""" |
| 160 | + raw = request.GET.get("page") |
| 161 | + try: |
| 162 | + return max(1, int(raw)) |
| 163 | + except (TypeError, ValueError): |
| 164 | + return 1 |
0 commit comments