pacnpal/thrillwiki_django_no_react
1
1
Fork 0
mirror of https://github.com/pacnpal/thrillwiki_django_no_react.git synced 2025-12-20 06:11:07 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
dcf890a55c6954f5c515b9190174001b407ee39a
thrillwiki_django_no_react/backend/apps/api/v1/viewsets.py
pacnpal dcf890a55c feat: Implement Entity Suggestion Manager and Modal components 
- Added EntitySuggestionManager.vue to manage entity suggestions and authentication.
- Created EntitySuggestionModal.vue for displaying suggestions and adding new entities.
- Integrated AuthManager for user authentication within the suggestion modal.
- Enhanced signal handling in start-servers.sh for graceful shutdown of servers.
- Improved server startup script to ensure proper cleanup and responsiveness to termination signals.
- Added documentation for signal handling fixes and usage instructions.
2025-08-25 10:46:54 -04:00

3143 lines
106 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 django.http import Http404
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":
# CRITICAL FIX: Check if this is a nested endpoint first
park_slug = self.kwargs.get("park_slug")
if park_slug:
# For nested endpoints, use the dedicated park selector
from apps.rides.selectors import rides_in_park
return rides_in_park(park_slug=park_slug)
# For global endpoints, parse filter parameters and use general selector
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:
try:
return ride_detail_optimized(slug=ride_slug, park_slug=park_slug)
except Ride.DoesNotExist:
raise Http404("Ride not found")
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]
# Optimized user lookup: single query using Q objects
from django.db.models import Q
from django.contrib.auth import get_user_model
User = get_user_model()
user = None
# Single query to find user by email OR username
try:
if "@" in email_or_username:
# Email-like input: try email first, then username as fallback
user_obj = (
User.objects.select_related()
.filter(
Q(email=email_or_username) | Q(username=email_or_username)
)
.first()
)
else:
# Username-like input: try username first, then email as fallback
user_obj = (
User.objects.select_related()
.filter(
Q(username=email_or_username) | Q(email=email_or_username)
)
.first()
)
if user_obj:
user = authenticate(
# type: ignore[attr-defined]
request._request,
username=user_obj.username,
password=password,
)
except Exception:
# Fallback to original behavior
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]
# Optimized token creation - get_or_create is atomic
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:
from django.core.cache import cache
from django.contrib.sites.shortcuts import get_current_site
site = get_current_site(request._request) # type: ignore[attr-defined]
# Cache key based on site and request host
cache_key = (
f"social_providers:{getattr(site, 'id', site.pk)}:{request.get_host()}"
)
# Try to get from cache first (cache for 15 minutes)
cached_providers = cache.get(cache_key)
if cached_providers is not None:
return Response(cached_providers)
providers_list = []
# Optimized query: filter by site and order by provider name
social_apps = SocialApp.objects.filter(sites=site).order_by("provider")
for social_app in social_apps:
try:
# Simplified provider name resolution - avoid expensive provider class loading
provider_name = social_app.name or social_app.provider.title()
# Build auth URL efficiently
auth_url = request.build_absolute_uri(
f"/accounts/{social_app.provider}/login/"
)
providers_list.append(
{
"id": social_app.provider,
"name": provider_name,
"authUrl": auth_url,
}
)
except Exception:
# Skip if provider can't be loaded
continue
# Serialize and cache the result
serializer = SocialProviderOutputSerializer(providers_list, many=True)
response_data = serializer.data
# Cache for 15 minutes (900 seconds)
cache.set(cache_key, response_data, 900)
return Response(response_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)
# === TRENDING VIEWSETS ===
@extend_schema_view(
list=extend_schema(
summary="Get trending content",
description="Retrieve trending parks and rides based on view counts, ratings, and recency.",
parameters=[
OpenApiParameter(
name="limit",
type=OpenApiTypes.INT,
location=OpenApiParameter.QUERY,
description="Number of trending items to return (default: 20, max: 100)",
),
OpenApiParameter(
name="timeframe",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Timeframe for trending calculation (day, week, month) - default: week",
),
],
responses={200: OpenApiTypes.OBJECT},
tags=["Trending"],
),
)
class TrendingAPIView(APIView):
"""API endpoint for trending content."""
permission_classes = [AllowAny]
def get(self, request: Request) -> Response:
"""Get trending parks and rides."""
from apps.core.services.trending_service import TrendingService
# Parse parameters
limit = min(int(request.query_params.get("limit", 20)), 100)
# Get trending content
trending_service = TrendingService()
all_trending = trending_service.get_trending_content(limit=limit * 2)
# Separate by content type
trending_rides = []
trending_parks = []
for item in all_trending:
if item.get("category") == "ride":
trending_rides.append(item)
elif item.get("category") == "park":
trending_parks.append(item)
# Limit each category
trending_rides = trending_rides[: limit // 3] if trending_rides else []
trending_parks = trending_parks[: limit // 3] if trending_parks else []
# Create mock latest reviews (since not implemented yet)
latest_reviews = [
{
"id": 1,
"name": "Steel Vengeance Review",
"location": "Cedar Point",
"category": "Roller Coaster",
"rating": 5.0,
"rank": 1,
"views": 1234,
"views_change": "+45%",
"slug": "steel-vengeance-review",
}
][: limit // 3]
# Return in expected frontend format
response_data = {
"trending_rides": trending_rides,
"trending_parks": trending_parks,
"latest_reviews": latest_reviews,
}
return Response(response_data)
@extend_schema_view(
list=extend_schema(
summary="Get new content",
description="Retrieve recently added parks and rides.",
parameters=[
OpenApiParameter(
name="limit",
type=OpenApiTypes.INT,
location=OpenApiParameter.QUERY,
description="Number of new items to return (default: 20, max: 100)",
),
OpenApiParameter(
name="days",
type=OpenApiTypes.INT,
location=OpenApiParameter.QUERY,
description="Number of days to look back for new content (default: 30, max: 365)",
),
],
responses={200: OpenApiTypes.OBJECT},
tags=["Trending"],
),
)
class NewContentAPIView(APIView):
"""API endpoint for new content."""
permission_classes = [AllowAny]
def get(self, request: Request) -> Response:
"""Get new parks and rides."""
from apps.core.services.trending_service import TrendingService
from datetime import datetime, date
# Parse parameters
limit = min(int(request.query_params.get("limit", 20)), 100)
# Get new content with longer timeframe to get more data
trending_service = TrendingService()
all_new_content = trending_service.get_new_content(
limit=limit * 2, days_back=60
)
recently_added = []
newly_opened = []
upcoming = []
# Categorize items based on date
today = date.today()
for item in all_new_content:
date_added = item.get("date_added", "")
if date_added:
try:
# Parse the date string
if isinstance(date_added, str):
item_date = datetime.fromisoformat(date_added).date()
else:
item_date = date_added
# Calculate days difference
days_diff = (today - item_date).days
if days_diff <= 30: # Recently added (last 30 days)
recently_added.append(item)
elif days_diff <= 365: # Newly opened (last year)
newly_opened.append(item)
else: # Older items
newly_opened.append(item)
except (ValueError, TypeError):
# If date parsing fails, add to recently added
recently_added.append(item)
else:
recently_added.append(item)
# Create mock upcoming items
upcoming = [
{
"id": 1,
"name": "Epic Universe",
"location": "Universal Orlando",
"category": "Theme Park",
"date_added": "Opening 2025",
"slug": "epic-universe",
},
{
"id": 2,
"name": "New Fantasyland Expansion",
"location": "Magic Kingdom",
"category": "Land Expansion",
"date_added": "Opening 2026",
"slug": "fantasyland-expansion",
},
]
# Limit each category
recently_added = recently_added[: limit // 3] if recently_added else []
newly_opened = newly_opened[: limit // 3] if newly_opened else []
upcoming = upcoming[: limit // 3] if upcoming else []
# Return in expected frontend format
response_data = {
"recently_added": recently_added,
"newly_opened": newly_opened,
"upcoming": upcoming,
}
return Response(response_data)
Reference in New Issue View Git Blame Copy Permalink