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