""" History API Views This module provides ViewSets for accessing historical data and change tracking across all models in the ThrillWiki system using django-pghistory. """ from drf_spectacular.utils import extend_schema, extend_schema_view, OpenApiParameter from drf_spectacular.types import OpenApiTypes from rest_framework.filters import OrderingFilter from rest_framework.permissions import AllowAny from rest_framework.response import Response from rest_framework.viewsets import ReadOnlyModelViewSet from rest_framework.request import Request from typing import Optional, cast, Sequence from django.shortcuts import get_object_or_404 from django.db.models import Count, QuerySet import pghistory.models from datetime import datetime # Import models from apps.parks.models import Park from apps.rides.models import Ride # Import serializers from .. import serializers as history_serializers from rest_framework import serializers as drf_serializers # Minimal fallback serializer used when a specific serializer symbol is missing. class _FallbackSerializer(drf_serializers.Serializer): def to_representation(self, instance): # return minimal safe representation so responses serialize without errors return {} ParkHistoryEventSerializer = getattr( history_serializers, "ParkHistoryEventSerializer", _FallbackSerializer ) RideHistoryEventSerializer = getattr( history_serializers, "RideHistoryEventSerializer", _FallbackSerializer ) ParkHistoryOutputSerializer = getattr( history_serializers, "ParkHistoryOutputSerializer", _FallbackSerializer ) RideHistoryOutputSerializer = getattr( history_serializers, "RideHistoryOutputSerializer", _FallbackSerializer ) UnifiedHistoryTimelineSerializer = getattr( history_serializers, "UnifiedHistoryTimelineSerializer", _FallbackSerializer ) # --- Constants for model strings to avoid duplication --- PARK_MODEL = "parks.park" RIDE_MODELS: Sequence[str] = [ "rides.ride", "rides.ridemodel", "rides.rollercoasterstats", ] COMPANY_MODELS: Sequence[str] = [ "companies.operator", "companies.propertyowner", "companies.manufacturer", "companies.designer", ] ACCOUNT_MODEL = "accounts.user" ALL_TRACKED_MODELS: Sequence[str] = [ PARK_MODEL, *RIDE_MODELS, *COMPANY_MODELS, ACCOUNT_MODEL, ] # --- Helper utilities to reduce duplicated logic / cognitive complexity --- def _parse_date(date_str: Optional[str]) -> Optional[datetime]: if not date_str: return None try: return datetime.strptime(date_str, "%Y-%m-%d") except ValueError: return None def _apply_list_filters( queryset: QuerySet, request: Request, *, default_limit: int = 50, max_limit: int = 500, ) -> QuerySet: """ Apply common 'list' filters: event_type, start/end date, and limit. Expects request to be a rest_framework.request.Request (cast by caller). """ # event_type event_type = request.query_params.get("event_type") if event_type == "created": queryset = queryset.filter(pgh_label="created") elif event_type == "updated": queryset = queryset.filter(pgh_label="updated") elif event_type == "deleted": queryset = queryset.filter(pgh_label="deleted") # date range start_date = _parse_date(request.query_params.get("start_date")) if start_date: queryset = queryset.filter(pgh_created_at__gte=start_date) end_date = _parse_date(request.query_params.get("end_date")) if end_date: queryset = queryset.filter(pgh_created_at__lte=end_date) # limit (slice the queryset) limit_raw = request.query_params.get("limit", str(default_limit)) try: limit_val = min(int(limit_raw), max_limit) queryset = queryset[:limit_val] except (ValueError, TypeError): queryset = queryset[:default_limit] return queryset @extend_schema_view( list=extend_schema( summary="Get park history", description="Retrieve history timeline for a specific park including all changes over time.", parameters=[ OpenApiParameter( name="limit", type=OpenApiTypes.INT, location=OpenApiParameter.QUERY, description="Number of history events to return (default: 50, max: 500)", ), OpenApiParameter( name="offset", type=OpenApiTypes.INT, location=OpenApiParameter.QUERY, description="Offset for pagination", ), OpenApiParameter( name="event_type", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, description="Filter by event type (created, updated, deleted)", ), OpenApiParameter( name="start_date", type=OpenApiTypes.DATE, location=OpenApiParameter.QUERY, description="Filter events after this date (YYYY-MM-DD)", ), OpenApiParameter( name="end_date", type=OpenApiTypes.DATE, location=OpenApiParameter.QUERY, description="Filter events before this date (YYYY-MM-DD)", ), ], responses={200: ParkHistoryEventSerializer(many=True)}, tags=["History", "Parks"], ), retrieve=extend_schema( summary="Get complete park history", description="Retrieve complete history for a park including current state and timeline.", responses={200: ParkHistoryOutputSerializer}, tags=["History", "Parks"], ), ) class ParkHistoryViewSet(ReadOnlyModelViewSet): """ ViewSet for accessing park history data. Provides read-only access to historical changes for parks, including version history and real-world changes. """ permission_classes = [AllowAny] lookup_field = "park_slug" filter_backends = [OrderingFilter] ordering_fields = ["pgh_created_at"] ordering = ["-pgh_created_at"] def get_queryset(self): # type: ignore[override] """Get history events for the specified park.""" park_slug = self.kwargs.get("park_slug") if not park_slug: return pghistory.models.Events.objects.none() # Get the park to ensure it exists park = get_object_or_404(Park, slug=park_slug) # Base queryset for park events queryset = ( pghistory.models.Events.objects.filter( pgh_model__in=[PARK_MODEL], pgh_obj_id=getattr(park, "id", None) ) .select_related() .order_by("-pgh_created_at") ) # Apply list filters via helper to reduce complexity if self.action == "list": queryset = _apply_list_filters( queryset, cast(Request, self.request), default_limit=50, max_limit=500 ) return queryset def get_serializer_class(self): # type: ignore[override] """Return appropriate serializer based on action.""" if self.action == "retrieve": return ParkHistoryOutputSerializer return ParkHistoryEventSerializer def retrieve(self, request, park_slug=None): """Get complete park history including current state.""" park = get_object_or_404(Park, slug=park_slug) # Get history events history_events = self.get_queryset()[:100] # Latest 100 events # safe attribute access using getattr to avoid static-checker complaints first_recorded = getattr(history_events.last(), "pgh_created_at", None) last_modified = getattr(history_events.first(), "pgh_created_at", None) # Prepare data for serializer history_data = { "park": park, "current_state": park, "summary": { "total_events": self.get_queryset().count(), "first_recorded": first_recorded, "last_modified": last_modified, }, "events": history_events, } serializer = ParkHistoryOutputSerializer(history_data) return Response(serializer.data) @extend_schema_view( list=extend_schema( summary="Get ride history", description="Retrieve history timeline for a specific ride including all changes over time.", parameters=[ OpenApiParameter( name="limit", type=OpenApiTypes.INT, location=OpenApiParameter.QUERY, description="Number of history events to return (default: 50, max: 500)", ), OpenApiParameter( name="offset", type=OpenApiTypes.INT, location=OpenApiParameter.QUERY, description="Offset for pagination", ), OpenApiParameter( name="event_type", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, description="Filter by event type (created, updated, deleted)", ), OpenApiParameter( name="start_date", type=OpenApiTypes.DATE, location=OpenApiParameter.QUERY, description="Filter events after this date (YYYY-MM-DD)", ), OpenApiParameter( name="end_date", type=OpenApiTypes.DATE, location=OpenApiParameter.QUERY, description="Filter events before this date (YYYY-MM-DD)", ), ], responses={200: RideHistoryEventSerializer(many=True)}, tags=["History", "Rides"], ), retrieve=extend_schema( summary="Get complete ride history", description="Retrieve complete history for a ride including current state and timeline.", responses={200: RideHistoryOutputSerializer}, tags=["History", "Rides"], ), ) class RideHistoryViewSet(ReadOnlyModelViewSet): """ ViewSet for accessing ride history data. Provides read-only access to historical changes for rides, including version history and real-world changes. """ permission_classes = [AllowAny] lookup_field = "ride_slug" filter_backends = [OrderingFilter] ordering_fields = ["pgh_created_at"] ordering = ["-pgh_created_at"] def get_queryset(self): # type: ignore[override] """Get history events for the specified ride.""" park_slug = self.kwargs.get("park_slug") ride_slug = self.kwargs.get("ride_slug") if not park_slug or not ride_slug: return pghistory.models.Events.objects.none() # Get the ride to ensure it exists ride = get_object_or_404(Ride, slug=ride_slug, park__slug=park_slug) # Base queryset for ride events queryset = ( pghistory.models.Events.objects.filter( pgh_model__in=RIDE_MODELS, pgh_obj_id=getattr(ride, "id", None) ) .select_related() .order_by("-pgh_created_at") ) # Apply list filters via helper if self.action == "list": queryset = _apply_list_filters( queryset, cast(Request, self.request), default_limit=50, max_limit=500 ) return queryset def get_serializer_class(self): # type: ignore[override] """Return appropriate serializer based on action.""" if self.action == "retrieve": return RideHistoryOutputSerializer return RideHistoryEventSerializer def retrieve(self, request, park_slug=None, ride_slug=None): """Get complete ride history including current state.""" ride = get_object_or_404(Ride, slug=ride_slug, park__slug=park_slug) # Get history events history_events = self.get_queryset()[:100] # Latest 100 events # safe attribute access first_recorded = getattr(history_events.last(), "pgh_created_at", None) last_modified = getattr(history_events.first(), "pgh_created_at", None) # Prepare data for serializer history_data = { "ride": ride, "current_state": ride, "summary": { "total_events": self.get_queryset().count(), "first_recorded": first_recorded, "last_modified": last_modified, }, "events": history_events, } serializer = RideHistoryOutputSerializer(history_data) return Response(serializer.data) @extend_schema_view( list=extend_schema( summary="Unified history timeline", description="Retrieve a unified timeline of all changes across parks, rides, and companies.", parameters=[ OpenApiParameter( name="limit", type=OpenApiTypes.INT, location=OpenApiParameter.QUERY, description="Number of history events to return (default: 100, max: 1000)", ), OpenApiParameter( name="offset", type=OpenApiTypes.INT, location=OpenApiParameter.QUERY, description="Offset for pagination", ), OpenApiParameter( name="model_type", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, description="Filter by model type (park, ride, company)", ), OpenApiParameter( name="event_type", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, description="Filter by event type (created, updated, deleted)", ), OpenApiParameter( name="start_date", type=OpenApiTypes.DATE, location=OpenApiParameter.QUERY, description="Filter events after this date (YYYY-MM-DD)", ), OpenApiParameter( name="end_date", type=OpenApiTypes.DATE, location=OpenApiParameter.QUERY, description="Filter events before this date (YYYY-MM-DD)", ), OpenApiParameter( name="significance", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, description="Filter by change significance (major, minor, routine)", ), ], responses={200: UnifiedHistoryTimelineSerializer}, tags=["History"], ), retrieve=extend_schema( summary="Get unified history timeline item", description="Retrieve a specific item from the unified history timeline.", responses={200: UnifiedHistoryTimelineSerializer}, tags=["History"], ), ) class UnifiedHistoryViewSet(ReadOnlyModelViewSet): """ ViewSet for unified history timeline across all models. Provides a comprehensive view of all changes across parks, rides, and companies in chronological order. """ permission_classes = [AllowAny] filter_backends = [OrderingFilter] ordering_fields = ["pgh_created_at"] ordering = ["-pgh_created_at"] def get_queryset(self): # type: ignore[override] """Get unified history events across all tracked models.""" queryset = ( pghistory.models.Events.objects.filter(pgh_model__in=ALL_TRACKED_MODELS) .select_related() .order_by("-pgh_created_at") ) # Filter by requested model_type (if provided) model_type = cast(Request, self.request).query_params.get("model_type") if model_type == "park": queryset = queryset.filter(pgh_model=PARK_MODEL) elif model_type == "ride": queryset = queryset.filter(pgh_model__in=RIDE_MODELS) elif model_type == "company": queryset = queryset.filter(pgh_model__in=COMPANY_MODELS) elif model_type == "user": queryset = queryset.filter(pgh_model=ACCOUNT_MODEL) # Apply shared list filters when serving the list action if self.action == "list": queryset = _apply_list_filters( queryset, cast(Request, self.request), default_limit=100, max_limit=1000 ) return queryset def get_serializer_class(self): # type: ignore[override] """Return unified history timeline serializer.""" return UnifiedHistoryTimelineSerializer def list(self, request): """Get unified history timeline with summary statistics.""" events = list(self.get_queryset()) # evaluate for counts / earliest/latest use # Summary statistics across all tracked models total_events = pghistory.models.Events.objects.filter( pgh_model__in=ALL_TRACKED_MODELS ).count() event_type_counts = ( pghistory.models.Events.objects.filter(pgh_model__in=ALL_TRACKED_MODELS) .values("pgh_label") .annotate(count=Count("id")) ) model_type_counts = ( pghistory.models.Events.objects.filter(pgh_model__in=ALL_TRACKED_MODELS) .values("pgh_model") .annotate(count=Count("id")) ) timeline_data = { "summary": { "total_events": total_events, "events_returned": len(events), "event_type_breakdown": { item["pgh_label"]: item["count"] for item in event_type_counts }, "model_type_breakdown": { item["pgh_model"]: item["count"] for item in model_type_counts }, "time_range": { "earliest": events[-1].pgh_created_at if events else None, "latest": events[0].pgh_created_at if events else None, }, }, "events": events, } serializer = UnifiedHistoryTimelineSerializer(timeline_data) return Response(serializer.data)