1from __future__ import annotations
  2
  3from typing import Any
  4
  5from plain import postgres
  6from plain.auth.views import AuthView
  7from plain.htmx.views import HTMXView
  8from plain.http import JsonResponse, Response
  9from plain.runtime import settings
 10from plain.urls import reverse
 11from plain.views import DetailView, ListView, RedirectView
 12
 13from .core import Observer
 14from .models import Trace
 15
 16
 17class ObserverIndexView(RedirectView):
 18    url_name = "observer:traces"
 19
 20
 21class ObserverTracesView(AuthView, HTMXView, ListView):
 22    template_name = "observer/traces.html"
 23    context_object_name = "traces"
 24    admin_required = True
 25
 26    def get_objects(self) -> postgres.QuerySet:
 27        return Trace.query.all()
 28
 29    def check_auth(self) -> None:
 30        # Allow the view if we're in DEBUG
 31        if settings.DEBUG:
 32            return
 33
 34        super().check_auth()
 35
 36    def after_response(self, response: Response) -> Response:
 37        response = super().after_response(response)
 38        # So we can load it in the toolbar
 39        response.headers["X-Frame-Options"] = "SAMEORIGIN"
 40        return response
 41
 42    def get_template_context(self) -> dict[str, Any]:
 43        context = super().get_template_context()
 44        context["observer"] = Observer.from_request(self.request)
 45        return context
 46
 47    def htmx_put_mode(self) -> Response:
 48        """Set observer mode via HTMX PUT."""
 49        mode = self.request.form_data.get("mode")
 50        observer = Observer.from_request(self.request)
 51
 52        response = Response(status_code=204)
 53        response.headers["HX-Refresh"] = "true"
 54
 55        if mode == "summary":
 56            observer.enable_summary_mode(response)
 57        elif mode == "persist":
 58            observer.enable_persist_mode(response)
 59        elif mode == "disable":
 60            observer.disable(response)
 61        else:
 62            return Response("Invalid mode", status_code=400)
 63
 64        return response
 65
 66    def htmx_delete_traces(self) -> Response:
 67        """Clear all traces via HTMX DELETE."""
 68        Trace.query.all().delete()
 69        response = Response(status_code=204)
 70        response.headers["HX-Refresh"] = "true"
 71        return response
 72
 73    def post(self) -> Response:
 74        """Handle POST requests to set observer mode."""
 75        action = self.request.form_data.get("observe_action")
 76        if action == "summary":
 77            observer = Observer.from_request(self.request)
 78            response = Response(status_code=204)
 79            observer.enable_summary_mode(response)
 80            return response
 81        elif action == "persist_once":
 82            observer = Observer.from_request(self.request)
 83            response = Response(status_code=204)
 84            observer.enable_persist_once_mode(response)
 85            return response
 86        return Response("Invalid action", status_code=400)
 87
 88
 89class ObserverTraceDetailView(AuthView, HTMXView, DetailView):
 90    """Detail view for a specific trace."""
 91
 92    template_name = "observer/trace_detail.html"
 93    context_object_name = "trace"
 94    admin_required = True
 95
 96    def get_object(self) -> Trace | None:
 97        return Trace.query.get_or_none(trace_id=self.url_kwargs.get("trace_id"))
 98
 99    def check_auth(self) -> None:
100        # Allow the view if we're in DEBUG
101        if settings.DEBUG:
102            return
103        super().check_auth()
104
105    def get(self) -> Response:
106        """Return trace data as HTML, JSON, or logs based on content negotiation."""
107        preferred = self.request.get_preferred_type("text/html", "application/json")
108        if (
109            preferred == "application/json"
110            or self.request.query_params.get("format") == "json"
111        ):
112            return JsonResponse(self.object.as_dict())
113
114        if self.request.query_params.get("logs") == "true":
115            logs = self.object.logs.query.all().order_by("timestamp")
116            log_lines = []
117            for log in logs:
118                timestamp = log.timestamp.strftime("%Y-%m-%d %H:%M:%S.%f")
119                log_lines.append(f"{timestamp} [{log.level}]: {log.message}")
120
121            return Response("\n".join(log_lines), content_type="text/plain")
122
123        return super().get()
124
125    def get_template_names(self) -> list[str]:
126        if self.is_htmx_request():
127            # Use a different template for HTMX requests
128            return ["observer/trace.html"]
129        return super().get_template_names()
130
131    def htmx_delete(self) -> Response:
132        self.object.delete()
133
134        # Redirect to traces list after deletion
135        response = Response(status_code=204)
136        response.headers["HX-Redirect"] = reverse("observer:traces")
137        return response