pacnpal/thrillwiki_django_no_react
1
1
Fork 0
mirror of https://github.com/pacnpal/thrillwiki_django_no_react.git synced 2025-12-20 08:51:09 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
e62646bcf938b1e8f2f6a802a8f77ac6492309b8
thrillwiki_django_no_react/backend/apps/api/v1/viewsets.py
pacnpal e62646bcf9 feat: major API restructure and Vue.js frontend integration 
- Centralize API endpoints in dedicated api app with v1 versioning
- Remove individual API modules from parks and rides apps
- Add event tracking system with analytics functionality
- Integrate Vue.js frontend with Tailwind CSS v4 and TypeScript
- Add comprehensive database migrations for event tracking
- Implement user authentication and social provider setup
- Add API schema documentation and serializers
- Configure development environment with shared scripts
- Update project structure for monorepo with frontend/backend separation
2025-08-24 16:42:20 -04:00

2911 lines
98 KiB
Python
Raw Blame History

"""
Consolidated ViewSets for ThrillWiki API v1.
This module consolidates all API ViewSets from different apps into a unified structure
following Django REST Framework and drf-spectacular best practices.
"""
import time
from django_filters.rest_framework import DjangoFilterBackend
from drf_spectacular.utils import extend_schema, extend_schema_view, OpenApiParameter
from drf_spectacular.types import OpenApiTypes
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.filters import SearchFilter, OrderingFilter
from rest_framework.permissions import (
IsAuthenticated,
IsAuthenticatedOrReadOnly,
AllowAny,
)
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet
from rest_framework.views import APIView
from rest_framework.authtoken.models import Token
from django.contrib.auth import authenticate, login, logout, get_user_model
from django.contrib.sites.shortcuts import get_current_site
from django.core.exceptions import ValidationError
from django.utils import timezone
from django.conf import settings
from django.shortcuts import get_object_or_404
from allauth.socialaccount.models import SocialApp
from allauth.socialaccount import providers
from health_check.views import MainView
import pghistory.models
# Import models from different apps
from apps.parks.models import Park, ParkArea, ParkLocation, ParkReview, Company
from apps.rides.models import (
Ride,
RideModel,
RollerCoasterStats,
RideLocation,
RideReview,
)
from apps.accounts.models import UserProfile, TopList, TopListItem
# Import selectors from different apps
from apps.parks.selectors import (
park_list_with_stats,
park_detail_optimized,
park_reviews_for_park,
park_statistics,
)
from apps.rides.selectors import (
ride_list_for_display,
ride_detail_optimized,
ride_statistics_by_category,
)
# Import services from different apps
from apps.parks.services import ParkService
# Import consolidated serializers
from .serializers import (
# Park serializers
ParkListOutputSerializer,
ParkDetailOutputSerializer,
ParkCreateInputSerializer,
ParkUpdateInputSerializer,
ParkFilterInputSerializer,
ParkStatsOutputSerializer,
ParkReviewOutputSerializer,
# Ride serializers
RideListOutputSerializer,
RideDetailOutputSerializer,
RideCreateInputSerializer,
RideUpdateInputSerializer,
RideFilterInputSerializer,
RideStatsOutputSerializer,
# Accounts serializers
UserOutputSerializer,
LoginInputSerializer,
LoginOutputSerializer,
SignupInputSerializer,
SignupOutputSerializer,
PasswordResetInputSerializer,
PasswordResetOutputSerializer,
PasswordChangeInputSerializer,
PasswordChangeOutputSerializer,
LogoutOutputSerializer,
SocialProviderOutputSerializer,
AuthStatusOutputSerializer,
# Health check serializers
HealthCheckOutputSerializer,
PerformanceMetricsOutputSerializer,
SimpleHealthOutputSerializer,
# History serializers
ParkHistoryEventSerializer,
RideHistoryEventSerializer,
ParkHistoryOutputSerializer,
RideHistoryOutputSerializer,
UnifiedHistoryTimelineSerializer,
# New comprehensive serializers
ParkAreaDetailOutputSerializer,
ParkAreaCreateInputSerializer,
ParkAreaUpdateInputSerializer,
ParkLocationOutputSerializer,
ParkLocationCreateInputSerializer,
ParkLocationUpdateInputSerializer,
CompanyDetailOutputSerializer,
CompanyCreateInputSerializer,
CompanyUpdateInputSerializer,
RideModelDetailOutputSerializer,
RideModelCreateInputSerializer,
RideModelUpdateInputSerializer,
RollerCoasterStatsOutputSerializer,
RollerCoasterStatsCreateInputSerializer,
RollerCoasterStatsUpdateInputSerializer,
RideLocationOutputSerializer,
RideLocationCreateInputSerializer,
RideLocationUpdateInputSerializer,
RideReviewOutputSerializer,
RideReviewCreateInputSerializer,
RideReviewUpdateInputSerializer,
UserProfileOutputSerializer,
UserProfileCreateInputSerializer,
UserProfileUpdateInputSerializer,
TopListOutputSerializer,
TopListCreateInputSerializer,
TopListUpdateInputSerializer,
TopListItemOutputSerializer,
TopListItemCreateInputSerializer,
TopListItemUpdateInputSerializer,
)
# Handle optional dependencies with fallback classes
class FallbackTurnstileMixin:
"""Fallback mixin if TurnstileMixin is not available."""
def validate_turnstile(self, request):
pass
class FallbackCacheMonitor:
"""Fallback class if CacheMonitor is not available."""
def get_cache_stats(self):
return {"error": "Cache monitoring not available"}
class FallbackIndexAnalyzer:
"""Fallback class if IndexAnalyzer is not available."""
@staticmethod
def analyze_slow_queries(threshold):
return {"error": "Query analysis not available"}
# Try to import the real classes, use fallbacks if not available
try:
from apps.accounts.mixins import TurnstileMixin
except ImportError:
TurnstileMixin = FallbackTurnstileMixin
try:
from apps.core.services.enhanced_cache_service import CacheMonitor
except ImportError:
CacheMonitor = FallbackCacheMonitor
try:
from apps.core.utils.query_optimization import IndexAnalyzer
except ImportError:
IndexAnalyzer = FallbackIndexAnalyzer
UserModel = get_user_model()
# === PARK VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List parks",
description="Retrieve a paginated list of theme parks with filtering and search capabilities.",
parameters=[
OpenApiParameter(
name="search",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Search parks by name or description",
),
OpenApiParameter(
name="status",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by park status (OPERATING, CLOSED_PERM, etc.)",
),
OpenApiParameter(
name="country",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by country",
),
OpenApiParameter(
name="state",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by state/province",
),
OpenApiParameter(
name="ordering",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Order results by field (name, opening_date, average_rating, etc.)",
),
],
responses={200: ParkListOutputSerializer(many=True)},
tags=["Parks"],
),
create=extend_schema(
summary="Create park",
description="Create a new theme park. Requires authentication.",
request=ParkCreateInputSerializer,
responses={
201: ParkDetailOutputSerializer,
400: OpenApiTypes.OBJECT,
401: OpenApiTypes.OBJECT,
},
tags=["Parks"],
),
retrieve=extend_schema(
summary="Get park details",
description="Retrieve detailed information about a specific park.",
responses={
200: ParkDetailOutputSerializer,
404: OpenApiTypes.OBJECT,
},
tags=["Parks"],
),
update=extend_schema(
summary="Update park",
description="Update a park's information. Requires authentication.",
request=ParkUpdateInputSerializer,
responses={
200: ParkDetailOutputSerializer,
400: OpenApiTypes.OBJECT,
401: OpenApiTypes.OBJECT,
404: OpenApiTypes.OBJECT,
},
tags=["Parks"],
),
partial_update=extend_schema(
summary="Partially update park",
description="Partially update a park's information. Requires authentication.",
request=ParkUpdateInputSerializer,
responses={
200: ParkDetailOutputSerializer,
400: OpenApiTypes.OBJECT,
401: OpenApiTypes.OBJECT,
404: OpenApiTypes.OBJECT,
},
tags=["Parks"],
),
destroy=extend_schema(
summary="Delete park",
description="Delete a park. Requires authentication and appropriate permissions.",
responses={
204: None,
401: OpenApiTypes.OBJECT,
403: OpenApiTypes.OBJECT,
404: OpenApiTypes.OBJECT,
},
tags=["Parks"],
),
stats=extend_schema(
summary="Get park statistics",
description="Retrieve global statistics about all parks in the system.",
responses={200: ParkStatsOutputSerializer},
tags=["Parks", "Statistics"],
),
reviews=extend_schema(
summary="Get park reviews",
description="Retrieve reviews for a specific park.",
responses={200: ParkReviewOutputSerializer(many=True)},
tags=["Parks", "Reviews"],
),
)
class ParkViewSet(ModelViewSet):
"""
ViewSet for managing theme parks.
Provides CRUD operations for parks plus additional endpoints for
statistics and reviews.
"""
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "slug"
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
search_fields = ["name", "description"]
ordering_fields = [
"name",
"opening_date",
"average_rating",
"coaster_count",
"created_at",
]
ordering = ["name"]
def get_queryset(self): # type: ignore[override]
"""Get optimized queryset based on action."""
if self.action == "list":
# Parse filter parameters for list view
filter_serializer = ParkFilterInputSerializer(
data=self.request.query_params # type: ignore[attr-defined]
)
filter_serializer.is_valid(raise_exception=True)
filters = filter_serializer.validated_data
return park_list_with_stats(filters=filters) # type: ignore[arg-type]
# For other actions, return base queryset
return Park.objects.select_related("operator", "property_owner").all()
def get_object(self): # type: ignore[override]
"""Get optimized object for detail operations."""
if self.action in ["retrieve", "update", "partial_update", "destroy"]:
slug = self.kwargs.get("slug")
return park_detail_optimized(slug=slug)
return super().get_object()
def get_serializer_class(self): # type: ignore[override]
"""Return appropriate serializer class based on action."""
if self.action == "list":
return ParkListOutputSerializer
elif self.action == "create":
return ParkCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return ParkUpdateInputSerializer
else:
return ParkDetailOutputSerializer
def perform_create(self, serializer):
"""Create park using service layer."""
park = ParkService.create_park(**serializer.validated_data)
serializer.instance = park
def perform_update(self, serializer):
"""Update park using service layer."""
park = ParkService.update_park(
park_id=self.get_object().id, **serializer.validated_data
)
serializer.instance = park
def perform_destroy(self, instance):
"""Delete park using service layer."""
ParkService.delete_park(park_id=instance.id)
@action(detail=False, methods=["get"])
def stats(self, request: Request) -> Response:
"""
Get park statistics.
Returns global statistics about all parks including totals,
averages, and top countries.
"""
stats = park_statistics()
serializer = ParkStatsOutputSerializer(stats)
return Response(
data=serializer.data,
headers={"Cache-Control": "max-age=3600"}, # 1 hour cache hint
)
@action(detail=True, methods=["get"])
def reviews(self, request: Request, slug: str | None = None) -> Response:
"""
Get reviews for a specific park.
Returns a list of user reviews for the park.
"""
park = self.get_object()
reviews = park_reviews_for_park(park_id=park.id, limit=50)
serializer = ParkReviewOutputSerializer(reviews, many=True)
return Response(
data=serializer.data,
headers={
"X-Total-Reviews": str(len(reviews)),
"X-Park-Name": park.name,
},
)
@action(detail=False, methods=["get"])
def recent_changes(self, request: Request) -> Response:
"""
Get recently changed parks.
Returns parks that have been modified recently with change details.
"""
days = request.query_params.get("days", "7")
try:
days = int(days)
except (ValueError, TypeError):
days = 7
# Get parks changed in the last N days
from datetime import timedelta
from django.utils import timezone
cutoff_date = timezone.now() - timedelta(days=days)
recent_events = (
pghistory.models.Events.objects.filter(
pgh_model="parks.park", pgh_created_at__gte=cutoff_date
)
.values("pgh_obj_id")
.distinct()
)
park_ids = [event["pgh_obj_id"] for event in recent_events]
changed_parks = Park.objects.filter(id__in=park_ids).select_related(
"operator", "property_owner"
)
serializer = ParkListOutputSerializer(changed_parks, many=True)
return Response(
{"count": len(changed_parks), "days": days, "parks": serializer.data}
)
@action(detail=False, methods=["get"])
def recent_openings(self, request: Request) -> Response:
"""
Get recently opened parks.
Returns parks that have opened in the specified time period.
"""
days = request.query_params.get("days", "30")
try:
days = int(days)
except (ValueError, TypeError):
days = 30
from datetime import timedelta
from django.utils import timezone
cutoff_date = timezone.now() - timedelta(days=days)
recent_openings = Park.objects.filter(
opening_date__gte=cutoff_date, status="OPERATING"
).select_related("operator", "property_owner")
serializer = ParkListOutputSerializer(recent_openings, many=True)
return Response(
{"count": len(recent_openings), "days": days, "parks": serializer.data}
)
@action(detail=False, methods=["get"])
def recent_closures(self, request: Request) -> Response:
"""
Get recently closed parks.
Returns parks that have closed or changed to non-operating status recently.
"""
days = request.query_params.get("days", "30")
try:
days = int(days)
except (ValueError, TypeError):
days = 30
from datetime import timedelta
from django.utils import timezone
cutoff_date = timezone.now() - timedelta(days=days)
# Get parks that have closure events in recent history
closure_events = (
pghistory.models.Events.objects.filter(
pgh_model="parks.park",
pgh_created_at__gte=cutoff_date,
pgh_data__contains={"status": "CLOSED_PERM"},
)
.values("pgh_obj_id")
.distinct()
)
park_ids = [event["pgh_obj_id"] for event in closure_events]
closed_parks = Park.objects.filter(id__in=park_ids).select_related(
"operator", "property_owner"
)
serializer = ParkListOutputSerializer(closed_parks, many=True)
return Response(
{"count": len(closed_parks), "days": days, "parks": serializer.data}
)
@action(detail=False, methods=["get"])
def recent_name_changes(self, request: Request) -> Response:
"""
Get parks with recent name changes.
Returns parks that have had their names changed recently.
"""
days = request.query_params.get("days", "90")
try:
days = int(days)
except (ValueError, TypeError):
days = 90
from datetime import timedelta
from django.utils import timezone
cutoff_date = timezone.now() - timedelta(days=days)
# Get parks with name change events
name_change_events = (
pghistory.models.Events.objects.filter(
pgh_model="parks.park",
pgh_created_at__gte=cutoff_date,
pgh_label="updated",
)
.values("pgh_obj_id")
.distinct()
)
park_ids = [event["pgh_obj_id"] for event in name_change_events]
changed_parks = Park.objects.filter(id__in=park_ids).select_related(
"operator", "property_owner"
)
serializer = ParkListOutputSerializer(changed_parks, many=True)
return Response(
{"count": len(changed_parks), "days": days, "parks": serializer.data}
)
# === RIDE VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List rides",
description="Retrieve a paginated list of rides with filtering and search capabilities.",
parameters=[
OpenApiParameter(
name="search",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Search rides by name or description",
),
OpenApiParameter(
name="category",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by ride category (RC, DR, FR, WR, TR, OT)",
),
OpenApiParameter(
name="status",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by ride status",
),
OpenApiParameter(
name="park_id",
type=OpenApiTypes.INT,
location=OpenApiParameter.QUERY,
description="Filter by park ID",
),
OpenApiParameter(
name="park_slug",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by park slug",
),
OpenApiParameter(
name="ordering",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Order results by field (name, opening_date, average_rating, etc.)",
),
],
responses={200: RideListOutputSerializer(many=True)},
tags=["Rides"],
),
create=extend_schema(
summary="Create ride",
description="Create a new ride. Requires authentication.",
request=RideCreateInputSerializer,
responses={
201: RideDetailOutputSerializer,
400: OpenApiTypes.OBJECT,
401: OpenApiTypes.OBJECT,
},
tags=["Rides"],
),
retrieve=extend_schema(
summary="Get ride details",
description="Retrieve detailed information about a specific ride.",
responses={
200: RideDetailOutputSerializer,
404: OpenApiTypes.OBJECT,
},
tags=["Rides"],
),
update=extend_schema(
summary="Update ride",
description="Update a ride's information. Requires authentication.",
request=RideUpdateInputSerializer,
responses={
200: RideDetailOutputSerializer,
400: OpenApiTypes.OBJECT,
401: OpenApiTypes.OBJECT,
404: OpenApiTypes.OBJECT,
},
tags=["Rides"],
),
partial_update=extend_schema(
summary="Partially update ride",
description="Partially update a ride's information. Requires authentication.",
request=RideUpdateInputSerializer,
responses={
200: RideDetailOutputSerializer,
400: OpenApiTypes.OBJECT,
401: OpenApiTypes.OBJECT,
404: OpenApiTypes.OBJECT,
},
tags=["Rides"],
),
destroy=extend_schema(
summary="Delete ride",
description="Delete a ride. Requires authentication and appropriate permissions.",
responses={
204: None,
401: OpenApiTypes.OBJECT,
403: OpenApiTypes.OBJECT,
404: OpenApiTypes.OBJECT,
},
tags=["Rides"],
),
stats=extend_schema(
summary="Get ride statistics",
description="Retrieve global statistics about all rides in the system.",
responses={200: RideStatsOutputSerializer},
tags=["Rides", "Statistics"],
),
)
class RideViewSet(ModelViewSet):
"""
ViewSet for managing rides.
Provides CRUD operations for rides plus additional endpoints for
statistics.
"""
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "slug"
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
search_fields = ["name", "description"]
ordering_fields = [
"name",
"opening_date",
"average_rating",
"capacity_per_hour",
"created_at",
]
ordering = ["name"]
def get_queryset(self): # type: ignore[override]
"""Get optimized queryset based on action."""
if self.action == "list":
# Parse filter parameters for list view
filter_serializer = RideFilterInputSerializer(
data=self.request.query_params # type: ignore[attr-defined]
)
filter_serializer.is_valid(raise_exception=True)
filters = filter_serializer.validated_data
return ride_list_for_display(filters=filters) # type: ignore[arg-type]
# For other actions, return base queryset
return Ride.objects.select_related(
"park", "park_area", "manufacturer", "designer", "ride_model"
).all()
def get_object(self): # type: ignore[override]
"""Get optimized object for detail operations."""
if self.action in ["retrieve", "update", "partial_update", "destroy"]:
# For rides, we need to get by park slug and ride slug
park_slug = self.kwargs.get("park_slug")
ride_slug = self.kwargs.get("slug") or self.kwargs.get("ride_slug")
if park_slug and ride_slug:
return ride_detail_optimized(slug=ride_slug, park_slug=park_slug)
elif ride_slug:
# For rides accessed directly by slug, we'll use the first approach
# and let the 404 handling work naturally
return super().get_object()
return super().get_object()
def get_serializer_class(self): # type: ignore[override]
"""Return appropriate serializer class based on action."""
if self.action == "list":
return RideListOutputSerializer
elif self.action == "create":
return RideCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return RideUpdateInputSerializer
else:
return RideDetailOutputSerializer
def perform_create(self, serializer):
"""Create ride using validated data."""
# For now, use standard Django creation
# TODO: Implement RideService for business logic
serializer.save()
def perform_update(self, serializer):
"""Update ride using validated data."""
# For now, use standard Django update
# TODO: Implement RideService for business logic
serializer.save()
def perform_destroy(self, instance):
"""Delete ride instance."""
# For now, use standard Django deletion
# TODO: Implement RideService for business logic
instance.delete()
@action(detail=False, methods=["get"])
def stats(self, request: Request) -> Response:
"""
Get ride statistics.
Returns global statistics about all rides including totals,
averages by category, and top manufacturers.
"""
# Import here to avoid circular imports
# Use the existing statistics function
stats = ride_statistics_by_category()
serializer = RideStatsOutputSerializer(stats)
return Response(
data=serializer.data,
headers={"Cache-Control": "max-age=3600"}, # 1 hour cache hint
)
@action(detail=False, methods=["get"])
def recent_changes(self, request: Request) -> Response:
"""
Get recently changed rides.
Returns rides that have been modified recently with change details.
"""
days = request.query_params.get("days", "7")
try:
days = int(days)
except (ValueError, TypeError):
days = 7
from datetime import timedelta
from django.utils import timezone
cutoff_date = timezone.now() - timedelta(days=days)
recent_events = (
pghistory.models.Events.objects.filter(
pgh_model__in=[
"rides.ride",
"rides.ridemodel",
"rides.rollercoasterstats",
],
pgh_created_at__gte=cutoff_date,
)
.values("pgh_obj_id")
.distinct()
)
ride_ids = [event["pgh_obj_id"] for event in recent_events]
changed_rides = Ride.objects.filter(id__in=ride_ids).select_related(
"park", "park_area", "manufacturer", "designer", "ride_model"
)
serializer = RideListOutputSerializer(changed_rides, many=True)
return Response(
{"count": len(changed_rides), "days": days, "rides": serializer.data}
)
@action(detail=False, methods=["get"])
def recent_openings(self, request: Request) -> Response:
"""
Get recently opened rides.
Returns rides that have opened in the specified time period.
"""
days = request.query_params.get("days", "30")
try:
days = int(days)
except (ValueError, TypeError):
days = 30
from datetime import timedelta
from django.utils import timezone
cutoff_date = timezone.now() - timedelta(days=days)
recent_openings = Ride.objects.filter(
opening_date__gte=cutoff_date, status="OPERATING"
).select_related("park", "park_area", "manufacturer", "designer", "ride_model")
serializer = RideListOutputSerializer(recent_openings, many=True)
return Response(
{"count": len(recent_openings), "days": days, "rides": serializer.data}
)
@action(detail=False, methods=["get"])
def recent_closures(self, request: Request) -> Response:
"""
Get recently closed rides.
Returns rides that have closed or changed to non-operating status recently.
"""
days = request.query_params.get("days", "30")
try:
days = int(days)
except (ValueError, TypeError):
days = 30
from datetime import timedelta
from django.utils import timezone
cutoff_date = timezone.now() - timedelta(days=days)
# Get rides that have closure events in recent history
closure_events = (
pghistory.models.Events.objects.filter(
pgh_model="rides.ride",
pgh_created_at__gte=cutoff_date,
pgh_data__contains={"status": "CLOSED_PERM"},
)
.values("pgh_obj_id")
.distinct()
)
ride_ids = [event["pgh_obj_id"] for event in closure_events]
closed_rides = Ride.objects.filter(id__in=ride_ids).select_related(
"park", "park_area", "manufacturer", "designer", "ride_model"
)
serializer = RideListOutputSerializer(closed_rides, many=True)
return Response(
{"count": len(closed_rides), "days": days, "rides": serializer.data}
)
@action(detail=False, methods=["get"])
def recent_name_changes(self, request: Request) -> Response:
"""
Get rides with recent name changes.
Returns rides that have had their names changed recently.
"""
days = request.query_params.get("days", "90")
try:
days = int(days)
except (ValueError, TypeError):
days = 90
from datetime import timedelta
from django.utils import timezone
cutoff_date = timezone.now() - timedelta(days=days)
# Get rides with name change events
name_change_events = (
pghistory.models.Events.objects.filter(
pgh_model="rides.ride",
pgh_created_at__gte=cutoff_date,
pgh_label="updated",
)
.values("pgh_obj_id")
.distinct()
)
ride_ids = [event["pgh_obj_id"] for event in name_change_events]
changed_rides = Ride.objects.filter(id__in=ride_ids).select_related(
"park", "park_area", "manufacturer", "designer", "ride_model"
)
serializer = RideListOutputSerializer(changed_rides, many=True)
return Response(
{"count": len(changed_rides), "days": days, "rides": serializer.data}
)
@action(detail=False, methods=["get"])
def recent_relocations(self, request: Request) -> Response:
"""
Get rides that have been relocated recently.
Returns rides that have moved between parks or areas recently.
"""
days = request.query_params.get(
"days", "365"
) # Default to 1 year for relocations
try:
days = int(days)
except (ValueError, TypeError):
days = 365
from datetime import timedelta
from django.utils import timezone
cutoff_date = timezone.now() - timedelta(days=days)
# Get rides with park/area change events
relocation_events = (
pghistory.models.Events.objects.filter(
pgh_model="rides.ride",
pgh_created_at__gte=cutoff_date,
pgh_label="updated",
)
.values("pgh_obj_id")
.distinct()
)
ride_ids = [event["pgh_obj_id"] for event in relocation_events]
relocated_rides = Ride.objects.filter(id__in=ride_ids).select_related(
"park", "park_area", "manufacturer", "designer", "ride_model"
)
serializer = RideListOutputSerializer(relocated_rides, many=True)
return Response(
{"count": len(relocated_rides), "days": days, "rides": serializer.data}
)
# === PARK AREA VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List park areas",
description="Retrieve a list of park areas with optional filtering",
responses={200: ParkAreaDetailOutputSerializer(many=True)},
tags=["Park Areas"],
),
create=extend_schema(
summary="Create park area",
description="Create a new park area",
request=ParkAreaCreateInputSerializer,
responses={201: ParkAreaDetailOutputSerializer},
tags=["Park Areas"],
),
retrieve=extend_schema(
summary="Get park area details",
description="Retrieve detailed information about a specific park area",
responses={200: ParkAreaDetailOutputSerializer},
tags=["Park Areas"],
),
update=extend_schema(
summary="Update park area",
description="Update park area information",
request=ParkAreaUpdateInputSerializer,
responses={200: ParkAreaDetailOutputSerializer},
tags=["Park Areas"],
),
destroy=extend_schema(
summary="Delete park area",
description="Delete a park area",
responses={204: None},
tags=["Park Areas"],
),
)
class ParkAreaViewSet(ModelViewSet):
"""ViewSet for managing park areas."""
queryset = ParkArea.objects.select_related("park").all()
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "id"
def get_serializer_class(self):
if self.action == "create":
return ParkAreaCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return ParkAreaUpdateInputSerializer
return ParkAreaDetailOutputSerializer
def perform_create(self, serializer):
park_id = serializer.validated_data.pop("park_id")
park = Park.objects.get(id=park_id)
serializer.save(park=park)
# === PARK LOCATION VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List park locations",
description="Retrieve a list of park locations",
responses={200: ParkLocationOutputSerializer(many=True)},
tags=["Park Locations"],
),
create=extend_schema(
summary="Create park location",
description="Create a new park location",
request=ParkLocationCreateInputSerializer,
responses={201: ParkLocationOutputSerializer},
tags=["Park Locations"],
),
retrieve=extend_schema(
summary="Get park location details",
description="Retrieve detailed information about a specific park location",
responses={200: ParkLocationOutputSerializer},
tags=["Park Locations"],
),
update=extend_schema(
summary="Update park location",
description="Update park location information",
request=ParkLocationUpdateInputSerializer,
responses={200: ParkLocationOutputSerializer},
tags=["Park Locations"],
),
destroy=extend_schema(
summary="Delete park location",
description="Delete a park location",
responses={204: None},
tags=["Park Locations"],
),
)
class ParkLocationViewSet(ModelViewSet):
"""ViewSet for managing park locations."""
queryset = ParkLocation.objects.select_related("park").all()
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "id"
def get_serializer_class(self):
if self.action == "create":
return ParkLocationCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return ParkLocationUpdateInputSerializer
return ParkLocationOutputSerializer
def perform_create(self, serializer):
park_id = serializer.validated_data.pop("park_id")
park = Park.objects.get(id=park_id)
serializer.save(park=park)
# === COMPANY VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List companies",
description="Retrieve a list of companies with optional role filtering",
parameters=[
OpenApiParameter(
name="roles",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by company roles (OPERATOR, MANUFACTURER, etc.)",
),
],
responses={200: CompanyDetailOutputSerializer(many=True)},
tags=["Companies"],
),
create=extend_schema(
summary="Create company",
description="Create a new company",
request=CompanyCreateInputSerializer,
responses={201: CompanyDetailOutputSerializer},
tags=["Companies"],
),
retrieve=extend_schema(
summary="Get company details",
description="Retrieve detailed information about a specific company",
responses={200: CompanyDetailOutputSerializer},
tags=["Companies"],
),
update=extend_schema(
summary="Update company",
description="Update company information",
request=CompanyUpdateInputSerializer,
responses={200: CompanyDetailOutputSerializer},
tags=["Companies"],
),
destroy=extend_schema(
summary="Delete company",
description="Delete a company",
responses={204: None},
tags=["Companies"],
),
)
class CompanyViewSet(ModelViewSet):
"""ViewSet for managing companies."""
queryset = Company.objects.all()
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "slug"
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
search_fields = ["name", "description"]
ordering_fields = ["name", "founded_date", "created_at"]
ordering = ["name"]
def get_queryset(self):
queryset = super().get_queryset()
roles = self.request.query_params.get("roles")
if roles:
role_list = roles.split(",")
queryset = queryset.filter(roles__overlap=role_list)
return queryset
def get_serializer_class(self):
if self.action == "create":
return CompanyCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return CompanyUpdateInputSerializer
return CompanyDetailOutputSerializer
# === RIDE MODEL VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List ride models",
description="Retrieve a list of ride models",
responses={200: RideModelDetailOutputSerializer(many=True)},
tags=["Ride Models"],
),
create=extend_schema(
summary="Create ride model",
description="Create a new ride model",
request=RideModelCreateInputSerializer,
responses={201: RideModelDetailOutputSerializer},
tags=["Ride Models"],
),
retrieve=extend_schema(
summary="Get ride model details",
description="Retrieve detailed information about a specific ride model",
responses={200: RideModelDetailOutputSerializer},
tags=["Ride Models"],
),
update=extend_schema(
summary="Update ride model",
description="Update ride model information",
request=RideModelUpdateInputSerializer,
responses={200: RideModelDetailOutputSerializer},
tags=["Ride Models"],
),
destroy=extend_schema(
summary="Delete ride model",
description="Delete a ride model",
responses={204: None},
tags=["Ride Models"],
),
)
class RideModelViewSet(ModelViewSet):
"""ViewSet for managing ride models."""
queryset = RideModel.objects.select_related("manufacturer").all()
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "id"
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
search_fields = ["name", "description"]
ordering_fields = ["name", "manufacturer__name", "created_at"]
ordering = ["manufacturer__name", "name"]
def get_serializer_class(self):
if self.action == "create":
return RideModelCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return RideModelUpdateInputSerializer
return RideModelDetailOutputSerializer
def perform_create(self, serializer):
manufacturer_id = serializer.validated_data.pop("manufacturer_id", None)
manufacturer = None
if manufacturer_id:
manufacturer = Company.objects.get(id=manufacturer_id)
serializer.save(manufacturer=manufacturer)
def perform_update(self, serializer):
manufacturer_id = serializer.validated_data.pop("manufacturer_id", None)
if manufacturer_id is not None:
manufacturer = (
Company.objects.get(id=manufacturer_id) if manufacturer_id else None
)
serializer.save(manufacturer=manufacturer)
else:
serializer.save()
# === ROLLER COASTER STATS VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List roller coaster stats",
description="Retrieve a list of roller coaster statistics",
responses={200: RollerCoasterStatsOutputSerializer(many=True)},
tags=["Roller Coaster Stats"],
),
create=extend_schema(
summary="Create roller coaster stats",
description="Create statistics for a roller coaster",
request=RollerCoasterStatsCreateInputSerializer,
responses={201: RollerCoasterStatsOutputSerializer},
tags=["Roller Coaster Stats"],
),
retrieve=extend_schema(
summary="Get roller coaster stats",
description="Retrieve statistics for a specific roller coaster",
responses={200: RollerCoasterStatsOutputSerializer},
tags=["Roller Coaster Stats"],
),
update=extend_schema(
summary="Update roller coaster stats",
description="Update roller coaster statistics",
request=RollerCoasterStatsUpdateInputSerializer,
responses={200: RollerCoasterStatsOutputSerializer},
tags=["Roller Coaster Stats"],
),
destroy=extend_schema(
summary="Delete roller coaster stats",
description="Delete roller coaster statistics",
responses={204: None},
tags=["Roller Coaster Stats"],
),
)
class RollerCoasterStatsViewSet(ModelViewSet):
"""ViewSet for managing roller coaster statistics."""
queryset = RollerCoasterStats.objects.select_related("ride", "ride__park").all()
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "id"
def get_serializer_class(self):
if self.action == "create":
return RollerCoasterStatsCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return RollerCoasterStatsUpdateInputSerializer
return RollerCoasterStatsOutputSerializer
def perform_create(self, serializer):
ride_id = serializer.validated_data.pop("ride_id")
ride = Ride.objects.get(id=ride_id)
serializer.save(ride=ride)
# === RIDE LOCATION VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List ride locations",
description="Retrieve a list of ride locations",
responses={200: RideLocationOutputSerializer(many=True)},
tags=["Ride Locations"],
),
create=extend_schema(
summary="Create ride location",
description="Create a location for a ride",
request=RideLocationCreateInputSerializer,
responses={201: RideLocationOutputSerializer},
tags=["Ride Locations"],
),
retrieve=extend_schema(
summary="Get ride location",
description="Retrieve location information for a specific ride",
responses={200: RideLocationOutputSerializer},
tags=["Ride Locations"],
),
update=extend_schema(
summary="Update ride location",
description="Update ride location information",
request=RideLocationUpdateInputSerializer,
responses={200: RideLocationOutputSerializer},
tags=["Ride Locations"],
),
destroy=extend_schema(
summary="Delete ride location",
description="Delete ride location",
responses={204: None},
tags=["Ride Locations"],
),
)
class RideLocationViewSet(ModelViewSet):
"""ViewSet for managing ride locations."""
queryset = RideLocation.objects.select_related("ride", "ride__park").all()
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "id"
def get_serializer_class(self):
if self.action == "create":
return RideLocationCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return RideLocationUpdateInputSerializer
return RideLocationOutputSerializer
def perform_create(self, serializer):
ride_id = serializer.validated_data.pop("ride_id")
ride = Ride.objects.get(id=ride_id)
serializer.save(ride=ride)
# === RIDE REVIEW VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List ride reviews",
description="Retrieve a list of ride reviews",
parameters=[
OpenApiParameter(
name="ride_id",
type=OpenApiTypes.INT,
location=OpenApiParameter.QUERY,
description="Filter by ride ID",
),
OpenApiParameter(
name="user",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by username",
),
],
responses={200: RideReviewOutputSerializer(many=True)},
tags=["Ride Reviews"],
),
create=extend_schema(
summary="Create ride review",
description="Create a new ride review",
request=RideReviewCreateInputSerializer,
responses={201: RideReviewOutputSerializer},
tags=["Ride Reviews"],
),
retrieve=extend_schema(
summary="Get ride review",
description="Retrieve a specific ride review",
responses={200: RideReviewOutputSerializer},
tags=["Ride Reviews"],
),
update=extend_schema(
summary="Update ride review",
description="Update a ride review (only by the author)",
request=RideReviewUpdateInputSerializer,
responses={200: RideReviewOutputSerializer},
tags=["Ride Reviews"],
),
destroy=extend_schema(
summary="Delete ride review",
description="Delete a ride review (only by the author)",
responses={204: None},
tags=["Ride Reviews"],
),
)
class RideReviewViewSet(ModelViewSet):
"""ViewSet for managing ride reviews."""
queryset = RideReview.objects.select_related("ride", "ride__park", "user").filter(
is_published=True
)
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "id"
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
search_fields = ["title", "content"]
ordering_fields = ["created_at", "rating", "visit_date"]
ordering = ["-created_at"]
def get_queryset(self):
queryset = super().get_queryset()
ride_id = self.request.query_params.get("ride_id")
user = self.request.query_params.get("user")
if ride_id:
queryset = queryset.filter(ride_id=ride_id)
if user:
queryset = queryset.filter(user__username=user)
return queryset
def get_serializer_class(self):
if self.action == "create":
return RideReviewCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return RideReviewUpdateInputSerializer
return RideReviewOutputSerializer
def perform_create(self, serializer):
ride_id = serializer.validated_data.pop("ride_id")
ride = Ride.objects.get(id=ride_id)
serializer.save(ride=ride, user=self.request.user)
def get_permissions(self):
"""
Instantiates and returns the list of permissions that this view requires.
"""
if self.action in ["create", "update", "partial_update", "destroy"]:
permission_classes = [IsAuthenticated]
else:
permission_classes = [AllowAny]
return [permission() for permission in permission_classes]
# === USER PROFILE VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List user profiles",
description="Retrieve a list of user profiles",
responses={200: UserProfileOutputSerializer(many=True)},
tags=["User Profiles"],
),
create=extend_schema(
summary="Create user profile",
description="Create a user profile",
request=UserProfileCreateInputSerializer,
responses={201: UserProfileOutputSerializer},
tags=["User Profiles"],
),
retrieve=extend_schema(
summary="Get user profile",
description="Retrieve a specific user profile",
responses={200: UserProfileOutputSerializer},
tags=["User Profiles"],
),
update=extend_schema(
summary="Update user profile",
description="Update user profile (only own profile)",
request=UserProfileUpdateInputSerializer,
responses={200: UserProfileOutputSerializer},
tags=["User Profiles"],
),
destroy=extend_schema(
summary="Delete user profile",
description="Delete user profile (only own profile)",
responses={204: None},
tags=["User Profiles"],
),
)
class UserProfileViewSet(ModelViewSet):
"""ViewSet for managing user profiles."""
queryset = UserProfile.objects.select_related("user").all()
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "profile_id"
filter_backends = [SearchFilter, OrderingFilter]
search_fields = ["display_name", "bio"]
ordering_fields = ["display_name", "coaster_credits"]
ordering = ["display_name"]
def get_serializer_class(self):
if self.action == "create":
return UserProfileCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return UserProfileUpdateInputSerializer
return UserProfileOutputSerializer
def perform_create(self, serializer):
serializer.save(user=self.request.user)
def get_permissions(self):
"""Only allow users to modify their own profiles."""
if self.action in ["create", "update", "partial_update", "destroy"]:
permission_classes = [IsAuthenticated]
else:
permission_classes = [AllowAny]
return [permission() for permission in permission_classes]
# === TOP LIST VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List top lists",
description="Retrieve a list of user top lists",
parameters=[
OpenApiParameter(
name="category",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by category (RC, DR, PK, etc.)",
),
OpenApiParameter(
name="user",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Filter by username",
),
],
responses={200: TopListOutputSerializer(many=True)},
tags=["Top Lists"],
),
create=extend_schema(
summary="Create top list",
description="Create a new top list",
request=TopListCreateInputSerializer,
responses={201: TopListOutputSerializer},
tags=["Top Lists"],
),
retrieve=extend_schema(
summary="Get top list",
description="Retrieve a specific top list",
responses={200: TopListOutputSerializer},
tags=["Top Lists"],
),
update=extend_schema(
summary="Update top list",
description="Update a top list (only by the owner)",
request=TopListUpdateInputSerializer,
responses={200: TopListOutputSerializer},
tags=["Top Lists"],
),
destroy=extend_schema(
summary="Delete top list",
description="Delete a top list (only by the owner)",
responses={204: None},
tags=["Top Lists"],
),
)
class TopListViewSet(ModelViewSet):
"""ViewSet for managing user top lists."""
queryset = TopList.objects.select_related("user").all()
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "id"
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
search_fields = ["title", "description"]
ordering_fields = ["title", "created_at", "updated_at"]
ordering = ["-updated_at"]
def get_queryset(self):
queryset = super().get_queryset()
category = self.request.query_params.get("category")
user = self.request.query_params.get("user")
if category:
queryset = queryset.filter(category=category)
if user:
queryset = queryset.filter(user__username=user)
return queryset
def get_serializer_class(self):
if self.action == "create":
return TopListCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return TopListUpdateInputSerializer
return TopListOutputSerializer
def perform_create(self, serializer):
serializer.save(user=self.request.user)
def get_permissions(self):
"""Allow authenticated users to create, but only owners can modify."""
if self.action in ["create", "update", "partial_update", "destroy"]:
permission_classes = [IsAuthenticated]
else:
permission_classes = [AllowAny]
return [permission() for permission in permission_classes]
# === TOP LIST ITEM VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="List top list items",
description="Retrieve items in top lists",
parameters=[
OpenApiParameter(
name="top_list_id",
type=OpenApiTypes.INT,
location=OpenApiParameter.QUERY,
description="Filter by top list ID",
),
],
responses={200: TopListItemOutputSerializer(many=True)},
tags=["Top List Items"],
),
create=extend_schema(
summary="Create top list item",
description="Add an item to a top list",
request=TopListItemCreateInputSerializer,
responses={201: TopListItemOutputSerializer},
tags=["Top List Items"],
),
retrieve=extend_schema(
summary="Get top list item",
description="Retrieve a specific top list item",
responses={200: TopListItemOutputSerializer},
tags=["Top List Items"],
),
update=extend_schema(
summary="Update top list item",
description="Update a top list item",
request=TopListItemUpdateInputSerializer,
responses={200: TopListItemOutputSerializer},
tags=["Top List Items"],
),
destroy=extend_schema(
summary="Delete top list item",
description="Remove an item from a top list",
responses={204: None},
tags=["Top List Items"],
),
)
class TopListItemViewSet(ModelViewSet):
"""ViewSet for managing top list items."""
queryset = TopListItem.objects.select_related(
"top_list", "top_list__user", "content_type"
).all()
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = "id"
ordering_fields = ["rank"]
ordering = ["rank"]
def get_queryset(self):
queryset = super().get_queryset()
top_list_id = self.request.query_params.get("top_list_id")
if top_list_id:
queryset = queryset.filter(top_list_id=top_list_id)
return queryset
def get_serializer_class(self):
if self.action == "create":
return TopListItemCreateInputSerializer
elif self.action in ["update", "partial_update"]:
return TopListItemUpdateInputSerializer
return TopListItemOutputSerializer
def perform_create(self, serializer):
top_list_id = serializer.validated_data.pop("top_list_id")
content_type_id = serializer.validated_data.pop("content_type_id")
object_id = serializer.validated_data.pop("object_id")
top_list = TopList.objects.get(id=top_list_id)
from django.contrib.contenttypes.models import ContentType
content_type = ContentType.objects.get(id=content_type_id)
serializer.save(
top_list=top_list,
content_type=content_type,
object_id=object_id,
)
def get_permissions(self):
"""Allow authenticated users to manage their own top list items."""
if self.action in ["create", "update", "partial_update", "destroy"]:
permission_classes = [IsAuthenticated]
else:
permission_classes = [AllowAny]
return [permission() for permission in permission_classes]
# === READ-ONLY VIEWSETS FOR REFERENCE DATA ===
class ParkReadOnlyViewSet(ReadOnlyModelViewSet):
"""
Read-only ViewSet for parks.
Provides list and retrieve operations for parks without
modification capabilities. Useful for reference data.
"""
queryset = Park.objects.select_related("operator", "property_owner").all()
serializer_class = ParkListOutputSerializer
lookup_field = "slug"
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
search_fields = ["name", "description"]
ordering_fields = ["name", "opening_date", "average_rating"]
ordering = ["name"]
def get_serializer_class(self): # type: ignore[override]
"""Return appropriate serializer class based on action."""
if self.action == "retrieve":
return ParkDetailOutputSerializer
return ParkListOutputSerializer
class RideReadOnlyViewSet(ReadOnlyModelViewSet):
"""
Read-only ViewSet for rides.
Provides list and retrieve operations for rides without
modification capabilities. Useful for reference data.
"""
queryset = Ride.objects.select_related(
"park", "park_area", "manufacturer", "designer", "ride_model"
).all()
serializer_class = RideListOutputSerializer
lookup_field = "slug"
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
search_fields = ["name", "description"]
ordering_fields = ["name", "opening_date", "average_rating"]
ordering = ["name"]
def get_serializer_class(self): # type: ignore[override]
"""Return appropriate serializer class based on action."""
if self.action == "retrieve":
return RideDetailOutputSerializer
return RideListOutputSerializer
# === ACCOUNTS VIEWSETS ===
@extend_schema_view(
post=extend_schema(
summary="User login",
description="Authenticate user with username/email and password.",
request=LoginInputSerializer,
responses={
200: LoginOutputSerializer,
400: OpenApiTypes.OBJECT,
},
tags=["Authentication"],
),
)
class LoginAPIView(TurnstileMixin, APIView):
"""API endpoint for user login."""
permission_classes = [AllowAny]
authentication_classes = []
serializer_class = LoginInputSerializer
def post(self, request: Request) -> Response:
try:
# Validate Turnstile if configured
self.validate_turnstile(request)
except ValidationError as e:
return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST)
serializer = LoginInputSerializer(data=request.data)
if serializer.is_valid():
# type: ignore[index]
email_or_username = serializer.validated_data["username"]
password = serializer.validated_data["password"] # type: ignore[index]
# Try to authenticate with email first, then username
user = None
if "@" in email_or_username:
try:
user_obj = UserModel.objects.get(email=email_or_username)
user = authenticate(
# type: ignore[attr-defined]
request._request,
username=user_obj.username,
password=password,
)
except UserModel.DoesNotExist:
pass
if not user:
user = authenticate(
# type: ignore[attr-defined]
request._request,
username=email_or_username,
password=password,
)
if user:
if user.is_active:
login(request._request, user) # type: ignore[attr-defined]
token, created = Token.objects.get_or_create(user=user)
response_serializer = LoginOutputSerializer(
{
"token": token.key,
"user": user,
"message": "Login successful",
}
)
return Response(response_serializer.data)
else:
return Response(
{"error": "Account is disabled"},
status=status.HTTP_400_BAD_REQUEST,
)
else:
return Response(
{"error": "Invalid credentials"},
status=status.HTTP_400_BAD_REQUEST,
)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
@extend_schema_view(
post=extend_schema(
summary="User registration",
description="Register a new user account.",
request=SignupInputSerializer,
responses={
201: SignupOutputSerializer,
400: OpenApiTypes.OBJECT,
},
tags=["Authentication"],
),
)
class SignupAPIView(TurnstileMixin, APIView):
"""API endpoint for user registration."""
permission_classes = [AllowAny]
authentication_classes = []
serializer_class = SignupInputSerializer
def post(self, request: Request) -> Response:
try:
# Validate Turnstile if configured
self.validate_turnstile(request)
except ValidationError as e:
return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST)
serializer = SignupInputSerializer(data=request.data)
if serializer.is_valid():
user = serializer.save()
login(request._request, user) # type: ignore[attr-defined]
token, created = Token.objects.get_or_create(user=user)
response_serializer = SignupOutputSerializer(
{
"token": token.key,
"user": user,
"message": "Registration successful",
}
)
return Response(response_serializer.data, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
@extend_schema_view(
post=extend_schema(
summary="User logout",
description="Logout the current user and invalidate their token.",
responses={
200: LogoutOutputSerializer,
401: OpenApiTypes.OBJECT,
},
tags=["Authentication"],
),
)
class LogoutAPIView(APIView):
"""API endpoint for user logout."""
permission_classes = [IsAuthenticated]
serializer_class = LogoutOutputSerializer
def post(self, request: Request) -> Response:
try:
# Delete the token for token-based auth
if hasattr(request.user, "auth_token"):
request.user.auth_token.delete()
# Logout from session
logout(request._request) # type: ignore[attr-defined]
response_serializer = LogoutOutputSerializer(
{"message": "Logout successful"}
)
return Response(response_serializer.data)
except Exception as e:
return Response(
{"error": "Logout failed"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
@extend_schema_view(
get=extend_schema(
summary="Get current user",
description="Retrieve information about the currently authenticated user.",
responses={
200: UserOutputSerializer,
401: OpenApiTypes.OBJECT,
},
tags=["Authentication"],
),
)
class CurrentUserAPIView(APIView):
"""API endpoint to get current user information."""
permission_classes = [IsAuthenticated]
serializer_class = UserOutputSerializer
def get(self, request: Request) -> Response:
serializer = UserOutputSerializer(request.user)
return Response(serializer.data)
@extend_schema_view(
post=extend_schema(
summary="Request password reset",
description="Send a password reset email to the user.",
request=PasswordResetInputSerializer,
responses={
200: PasswordResetOutputSerializer,
400: OpenApiTypes.OBJECT,
},
tags=["Authentication"],
),
)
class PasswordResetAPIView(APIView):
"""API endpoint to request password reset."""
permission_classes = [AllowAny]
serializer_class = PasswordResetInputSerializer
def post(self, request: Request) -> Response:
serializer = PasswordResetInputSerializer(
data=request.data, context={"request": request}
)
if serializer.is_valid():
serializer.save()
response_serializer = PasswordResetOutputSerializer(
{"detail": "Password reset email sent"}
)
return Response(response_serializer.data)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
@extend_schema_view(
post=extend_schema(
summary="Change password",
description="Change the current user's password.",
request=PasswordChangeInputSerializer,
responses={
200: PasswordChangeOutputSerializer,
400: OpenApiTypes.OBJECT,
401: OpenApiTypes.OBJECT,
},
tags=["Authentication"],
),
)
class PasswordChangeAPIView(APIView):
"""API endpoint to change password."""
permission_classes = [IsAuthenticated]
serializer_class = PasswordChangeInputSerializer
def post(self, request: Request) -> Response:
serializer = PasswordChangeInputSerializer(
data=request.data, context={"request": request}
)
if serializer.is_valid():
serializer.save()
response_serializer = PasswordChangeOutputSerializer(
{"detail": "Password changed successfully"}
)
return Response(response_serializer.data)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
@extend_schema_view(
get=extend_schema(
summary="Get social providers",
description="Retrieve available social authentication providers.",
responses={200: SocialProviderOutputSerializer(many=True)},
tags=["Authentication"],
),
)
class SocialProvidersAPIView(APIView):
"""API endpoint to get available social authentication providers."""
permission_classes = [AllowAny]
serializer_class = SocialProviderOutputSerializer
def get(self, request: Request) -> Response:
site = get_current_site(request._request) # type: ignore[attr-defined]
providers_list = []
# Get all configured social apps for the current site
social_apps = SocialApp.objects.filter(sites=site)
for social_app in social_apps:
try:
# Get provider class from providers module
provider_module = getattr(providers, social_app.provider, None)
if provider_module and hasattr(provider_module, "provider"):
provider_class = provider_module.provider
provider_instance = provider_class(request)
auth_url = request.build_absolute_uri(
f"/accounts/{social_app.provider}/login/"
)
providers_list.append(
{
"id": social_app.provider,
"name": provider_instance.name,
"authUrl": auth_url,
}
)
else:
# Fallback: use provider id as name
auth_url = request.build_absolute_uri(
f"/accounts/{social_app.provider}/login/"
)
providers_list.append(
{
"id": social_app.provider,
"name": social_app.provider.title(),
"authUrl": auth_url,
}
)
except Exception:
# Skip if provider can't be loaded
continue
serializer = SocialProviderOutputSerializer(providers_list, many=True)
return Response(serializer.data)
@extend_schema_view(
post=extend_schema(
summary="Check authentication status",
description="Check if user is authenticated and return user data.",
responses={200: AuthStatusOutputSerializer},
tags=["Authentication"],
),
)
class AuthStatusAPIView(APIView):
"""API endpoint to check authentication status."""
permission_classes = [AllowAny]
serializer_class = AuthStatusOutputSerializer
def post(self, request: Request) -> Response:
if request.user.is_authenticated:
response_data = {
"authenticated": True,
"user": request.user,
}
else:
response_data = {
"authenticated": False,
"user": None,
}
serializer = AuthStatusOutputSerializer(response_data)
return Response(serializer.data)
# === HEALTH CHECK VIEWSETS ===
@extend_schema_view(
get=extend_schema(
summary="Health check",
description="Get comprehensive health check information including system metrics.",
responses={
200: HealthCheckOutputSerializer,
503: HealthCheckOutputSerializer,
},
tags=["Health"],
),
)
class HealthCheckAPIView(APIView):
"""Enhanced API endpoint for health checks with detailed JSON response."""
permission_classes = [AllowAny]
serializer_class = HealthCheckOutputSerializer
def get(self, request: Request) -> Response:
"""Return comprehensive health check information."""
start_time = time.time()
# Get basic health check results
main_view = MainView()
main_view.request = request._request # type: ignore[attr-defined]
plugins = main_view.plugins
errors = main_view.errors
# Collect additional performance metrics
try:
cache_monitor = CacheMonitor()
cache_stats = cache_monitor.get_cache_stats()
except Exception:
cache_stats = {"error": "Cache monitoring unavailable"}
# Build comprehensive health data
health_data = {
"status": "healthy" if not errors else "unhealthy",
"timestamp": timezone.now(),
"version": getattr(settings, "VERSION", "1.0.0"),
"environment": getattr(settings, "ENVIRONMENT", "development"),
"response_time_ms": 0, # Will be calculated at the end
"checks": {},
"metrics": {
"cache": cache_stats,
"database": self._get_database_metrics(),
"system": self._get_system_metrics(),
},
}
# Process individual health checks
for plugin in plugins:
plugin_name = plugin.identifier()
plugin_errors = (
errors.get(plugin.__class__.__name__, [])
if isinstance(errors, dict)
else []
)
health_data["checks"][plugin_name] = {
"status": "healthy" if not plugin_errors else "unhealthy",
"critical": getattr(plugin, "critical_service", False),
"errors": [str(error) for error in plugin_errors],
"response_time_ms": getattr(plugin, "_response_time", None),
}
# Calculate total response time
health_data["response_time_ms"] = round((time.time() - start_time) * 1000, 2)
# Determine HTTP status code
status_code = 200
if errors:
# Check if any critical services are failing
critical_errors = any(
getattr(plugin, "critical_service", False)
for plugin in plugins
if isinstance(errors, dict) and errors.get(plugin.__class__.__name__)
)
status_code = 503 if critical_errors else 200
serializer = HealthCheckOutputSerializer(health_data)
return Response(serializer.data, status=status_code)
def _get_database_metrics(self):
"""Get database performance metrics."""
try:
from django.db import connection
# Get basic connection info
metrics = {
"vendor": connection.vendor,
"connection_status": "connected",
}
# Test query performance
start_time = time.time()
with connection.cursor() as cursor:
cursor.execute("SELECT 1")
cursor.fetchone()
query_time = (time.time() - start_time) * 1000
metrics["test_query_time_ms"] = round(query_time, 2)
# PostgreSQL specific metrics
if connection.vendor == "postgresql":
try:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT
numbackends as active_connections,
xact_commit as transactions_committed,
xact_rollback as transactions_rolled_back,
blks_read as blocks_read,
blks_hit as blocks_hit
FROM pg_stat_database
WHERE datname = current_database()
"""
)
row = cursor.fetchone()
if row:
metrics.update(
{ # type: ignore[arg-type]
"active_connections": row[0],
"transactions_committed": row[1],
"transactions_rolled_back": row[2],
"cache_hit_ratio": (
round((row[4] / (row[3] + row[4])) * 100, 2)
if (row[3] + row[4]) > 0
else 0
),
}
)
except Exception:
pass # Skip advanced metrics if not available
return metrics
except Exception as e:
return {"connection_status": "error", "error": str(e)}
def _get_system_metrics(self):
"""Get system performance metrics."""
metrics = {
"debug_mode": settings.DEBUG,
"allowed_hosts": (settings.ALLOWED_HOSTS if settings.DEBUG else ["hidden"]),
}
try:
import psutil
# Memory metrics
memory = psutil.virtual_memory()
metrics["memory"] = {
"total_mb": round(memory.total / 1024 / 1024, 2),
"available_mb": round(memory.available / 1024 / 1024, 2),
"percent_used": memory.percent,
}
# CPU metrics
metrics["cpu"] = {
"percent_used": psutil.cpu_percent(interval=0.1),
"core_count": psutil.cpu_count(),
}
# Disk metrics
disk = psutil.disk_usage("/")
metrics["disk"] = {
"total_gb": round(disk.total / 1024 / 1024 / 1024, 2),
"free_gb": round(disk.free / 1024 / 1024 / 1024, 2),
"percent_used": round((disk.used / disk.total) * 100, 2),
}
except ImportError:
metrics["system_monitoring"] = "psutil not available"
except Exception as e:
metrics["system_error"] = str(e)
return metrics
@extend_schema_view(
get=extend_schema(
summary="Performance metrics",
description="Get performance metrics and database analysis (debug mode only).",
responses={
200: PerformanceMetricsOutputSerializer,
403: OpenApiTypes.OBJECT,
},
tags=["Health"],
),
)
class PerformanceMetricsAPIView(APIView):
"""API view for performance metrics and database analysis."""
permission_classes = [AllowAny] if settings.DEBUG else []
serializer_class = PerformanceMetricsOutputSerializer
def get(self, request: Request) -> Response:
"""Return performance metrics and analysis."""
if not settings.DEBUG:
return Response({"error": "Only available in debug mode"}, status=403)
metrics = {
"timestamp": timezone.now(),
"database_analysis": self._get_database_analysis(),
"cache_performance": self._get_cache_performance(),
"recent_slow_queries": self._get_slow_queries(),
}
serializer = PerformanceMetricsOutputSerializer(metrics)
return Response(serializer.data)
def _get_database_analysis(self):
"""Analyze database performance."""
try:
from django.db import connection
analysis = {
"total_queries": len(connection.queries),
"query_analysis": IndexAnalyzer.analyze_slow_queries(0.05),
}
if connection.queries:
query_times = [float(q.get("time", 0)) for q in connection.queries]
analysis.update(
{
"total_query_time": sum(query_times),
"average_query_time": sum(query_times) / len(query_times),
"slowest_query_time": max(query_times),
"fastest_query_time": min(query_times),
}
)
return analysis
except Exception as e:
return {"error": str(e)}
def _get_cache_performance(self):
"""Get cache performance metrics."""
try:
cache_monitor = CacheMonitor()
return cache_monitor.get_cache_stats()
except Exception as e:
return {"error": str(e)}
def _get_slow_queries(self):
"""Get recent slow queries."""
try:
return IndexAnalyzer.analyze_slow_queries(0.1) # 100ms threshold
except Exception as e:
return {"error": str(e)}
@extend_schema_view(
get=extend_schema(
summary="Simple health check",
description="Simple health check endpoint for load balancers.",
responses={
200: SimpleHealthOutputSerializer,
503: SimpleHealthOutputSerializer,
},
tags=["Health"],
),
)
class SimpleHealthAPIView(APIView):
"""Simple health check endpoint for load balancers."""
permission_classes = [AllowAny]
serializer_class = SimpleHealthOutputSerializer
def get(self, request: Request) -> Response:
"""Return simple OK status."""
try:
# Basic database connectivity test
from django.db import connection
with connection.cursor() as cursor:
cursor.execute("SELECT 1")
cursor.fetchone()
response_data = {
"status": "ok",
"timestamp": timezone.now(),
}
serializer = SimpleHealthOutputSerializer(response_data)
return Response(serializer.data)
except Exception as e:
response_data = {
"status": "error",
"error": str(e),
"timestamp": timezone.now(),
}
serializer = SimpleHealthOutputSerializer(response_data)
return Response(serializer.data, status=503)
# === HISTORY VIEWSETS ===
@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):
"""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)
# Get all history events for this park
queryset = (
pghistory.models.Events.objects.filter(
pgh_model__in=["parks.park"], pgh_obj_id=park.id
)
.select_related()
.order_by("-pgh_created_at")
)
# Apply filters
if self.action == "list":
# Filter by event type
event_type = self.request.query_params.get("event_type")
if 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")
# Filter by date range
start_date = self.request.query_params.get("start_date")
if start_date:
try:
from datetime import datetime
start_datetime = datetime.strptime(start_date, "%Y-%m-%d")
queryset = queryset.filter(pgh_created_at__gte=start_datetime)
except ValueError:
pass
end_date = self.request.query_params.get("end_date")
if end_date:
try:
from datetime import datetime
end_datetime = datetime.strptime(end_date, "%Y-%m-%d")
queryset = queryset.filter(pgh_created_at__lte=end_datetime)
except ValueError:
pass
# Apply limit
limit = self.request.query_params.get("limit", "50")
try:
limit = min(int(limit), 500) # Max 500 events
queryset = queryset[:limit]
except (ValueError, TypeError):
queryset = queryset[:50]
return queryset
def get_serializer_class(self):
"""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
# Prepare data for serializer
history_data = {
"park": park,
"current_state": park,
"summary": {
"total_events": self.get_queryset().count(),
"first_recorded": (
history_events.last().pgh_created_at if history_events else None
),
"last_modified": (
history_events.first().pgh_created_at if history_events else None
),
},
"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):
"""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)
# Get all history events for this ride
queryset = (
pghistory.models.Events.objects.filter(
pgh_model__in=[
"rides.ride",
"rides.ridemodel",
"rides.rollercoasterstats",
],
pgh_obj_id=ride.id,
)
.select_related()
.order_by("-pgh_created_at")
)
# Apply the same filtering logic as ParkHistoryViewSet
if self.action == "list":
# Filter by event type
event_type = self.request.query_params.get("event_type")
if 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")
# Filter by date range
start_date = self.request.query_params.get("start_date")
if start_date:
try:
from datetime import datetime
start_datetime = datetime.strptime(start_date, "%Y-%m-%d")
queryset = queryset.filter(pgh_created_at__gte=start_datetime)
except ValueError:
pass
end_date = self.request.query_params.get("end_date")
if end_date:
try:
from datetime import datetime
end_datetime = datetime.strptime(end_date, "%Y-%m-%d")
queryset = queryset.filter(pgh_created_at__lte=end_datetime)
except ValueError:
pass
# Apply limit
limit = self.request.query_params.get("limit", "50")
try:
limit = min(int(limit), 500) # Max 500 events
queryset = queryset[:limit]
except (ValueError, TypeError):
queryset = queryset[:50]
return queryset
def get_serializer_class(self):
"""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
# Prepare data for serializer
history_data = {
"ride": ride,
"current_state": ride,
"summary": {
"total_events": self.get_queryset().count(),
"first_recorded": (
history_events.last().pgh_created_at if history_events else None
),
"last_modified": (
history_events.first().pgh_created_at if history_events else None
),
},
"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"],
),
)
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):
"""Get unified history events across all tracked models."""
queryset = (
pghistory.models.Events.objects.filter(
pgh_model__in=[
"parks.park",
"rides.ride",
"rides.ridemodel",
"rides.rollercoasterstats",
"companies.operator",
"companies.propertyowner",
"companies.manufacturer",
"companies.designer",
"accounts.user",
]
)
.select_related()
.order_by("-pgh_created_at")
)
# Apply filters
model_type = self.request.query_params.get("model_type")
if model_type:
if model_type == "park":
queryset = queryset.filter(pgh_model="parks.park")
elif model_type == "ride":
queryset = queryset.filter(
pgh_model__in=[
"rides.ride",
"rides.ridemodel",
"rides.rollercoasterstats",
]
)
elif model_type == "company":
queryset = queryset.filter(
pgh_model__in=[
"companies.operator",
"companies.propertyowner",
"companies.manufacturer",
"companies.designer",
]
)
elif model_type == "user":
queryset = queryset.filter(pgh_model="accounts.user")
# Filter by event type
event_type = self.request.query_params.get("event_type")
if 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")
# Filter by date range
start_date = self.request.query_params.get("start_date")
if start_date:
try:
from datetime import datetime
start_datetime = datetime.strptime(start_date, "%Y-%m-%d")
queryset = queryset.filter(pgh_created_at__gte=start_datetime)
except ValueError:
pass
end_date = self.request.query_params.get("end_date")
if end_date:
try:
from datetime import datetime
end_datetime = datetime.strptime(end_date, "%Y-%m-%d")
queryset = queryset.filter(pgh_created_at__lte=end_datetime)
except ValueError:
pass
# Apply limit
limit = self.request.query_params.get("limit", "100")
try:
limit = min(int(limit), 1000) # Max 1000 events
queryset = queryset[:limit]
except (ValueError, TypeError):
queryset = queryset[:100]
return queryset
def get_serializer_class(self):
"""Return unified history timeline serializer."""
return UnifiedHistoryTimelineSerializer
def list(self, request):
"""Get unified history timeline with summary statistics."""
events = self.get_queryset()
# Calculate summary statistics
total_events = pghistory.models.Events.objects.filter(
pgh_model__in=[
"parks.park",
"rides.ride",
"rides.ridemodel",
"rides.rollercoasterstats",
"companies.operator",
"companies.propertyowner",
"companies.manufacturer",
"companies.designer",
"accounts.user",
]
).count()
# Get event type counts
from django.db.models import Count
event_type_counts = (
pghistory.models.Events.objects.filter(
pgh_model__in=[
"parks.park",
"rides.ride",
"rides.ridemodel",
"rides.rollercoasterstats",
"companies.operator",
"companies.propertyowner",
"companies.manufacturer",
"companies.designer",
"accounts.user",
]
)
.values("pgh_label")
.annotate(count=Count("id"))
)
# Get model type counts
model_type_counts = (
pghistory.models.Events.objects.filter(
pgh_model__in=[
"parks.park",
"rides.ride",
"rides.ridemodel",
"rides.rollercoasterstats",
"companies.operator",
"companies.propertyowner",
"companies.manufacturer",
"companies.designer",
"accounts.user",
]
)
.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.last().pgh_created_at if events else None,
"latest": events.first().pgh_created_at if events else None,
},
},
"events": events,
}
serializer = UnifiedHistoryTimelineSerializer(timeline_data)
return Response(serializer.data)
Reference in New Issue View Git Blame Copy Permalink