pacnpal/thrillwiki_django_no_react
1
1
Fork 0
mirror of https://github.com/pacnpal/thrillwiki_django_no_react.git synced 2025-12-23 02:31:08 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add comprehensive tests for Parks API and models

Browse Source 
- Implemented extensive test cases for the Parks API, covering endpoints for listing, retrieving, creating, updating, and deleting parks.
- Added tests for filtering, searching, and ordering parks in the API.
- Created tests for error handling in the API, including malformed JSON and unsupported methods.
- Developed model tests for Park, ParkArea, Company, and ParkReview models, ensuring validation and constraints are enforced.
- Introduced utility mixins for API and model testing to streamline assertions and enhance test readability.
- Included integration tests to validate complete workflows involving park creation, retrieval, updating, and deletion.
This commit is contained in:
pacnpal
2025-08-17 19:36:20 -04:00
parent 17228e9935
commit c26414ff74
210 changed files with 24155 additions and 833 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
parks/api/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Parks API module

295
parks/api/serializers.py Normal file
View File

@@ -0,0 +1,295 @@
"""
Serializers for Parks API following Django styleguide patterns.
Separates Input and Output serializers for clear boundaries.
"""
from rest_framework import serializers
from django.contrib.gis.geos import Point
from ..models import Park, ParkArea, Company, ParkReview
class ParkLocationOutputSerializer(serializers.Serializer):
"""Output serializer for park location data."""
latitude = serializers.SerializerMethodField()
longitude = serializers.SerializerMethodField()
city = serializers.SerializerMethodField()
state = serializers.SerializerMethodField()
country = serializers.SerializerMethodField()
formatted_address = serializers.SerializerMethodField()
def get_latitude(self, obj):
if hasattr(obj, 'location') and obj.location:
return obj.location.latitude
return None
def get_longitude(self, obj):
if hasattr(obj, 'location') and obj.location:
return obj.location.longitude
return None
def get_city(self, obj):
if hasattr(obj, 'location') and obj.location:
return obj.location.city
return None
def get_state(self, obj):
if hasattr(obj, 'location') and obj.location:
return obj.location.state
return None
def get_country(self, obj):
if hasattr(obj, 'location') and obj.location:
return obj.location.country
return None
def get_formatted_address(self, obj):
if hasattr(obj, 'location') and obj.location:
return obj.location.formatted_address
return ""
class CompanyOutputSerializer(serializers.Serializer):
"""Output serializer for company data."""
id = serializers.IntegerField()
name = serializers.CharField()
slug = serializers.CharField()
roles = serializers.ListField(child=serializers.CharField())
class ParkAreaOutputSerializer(serializers.Serializer):
"""Output serializer for park area data."""
id = serializers.IntegerField()
name = serializers.CharField()
slug = serializers.CharField()
description = serializers.CharField()
class ParkListOutputSerializer(serializers.Serializer):
"""Output serializer for park list view."""
id = serializers.IntegerField()
name = serializers.CharField()
slug = serializers.CharField()
status = serializers.CharField()
description = serializers.CharField()
# Statistics
average_rating = serializers.DecimalField(max_digits=3, decimal_places=2, allow_null=True)
coaster_count = serializers.IntegerField(allow_null=True)
ride_count = serializers.IntegerField(allow_null=True)
# Location (simplified for list view)
location = ParkLocationOutputSerializer(allow_null=True)
# Operator info
operator = CompanyOutputSerializer()
# Metadata
created_at = serializers.DateTimeField()
updated_at = serializers.DateTimeField()
class ParkDetailOutputSerializer(serializers.Serializer):
"""Output serializer for park detail view."""
id = serializers.IntegerField()
name = serializers.CharField()
slug = serializers.CharField()
status = serializers.CharField()
description = serializers.CharField()
# Details
opening_date = serializers.DateField(allow_null=True)
closing_date = serializers.DateField(allow_null=True)
operating_season = serializers.CharField()
size_acres = serializers.DecimalField(max_digits=10, decimal_places=2, allow_null=True)
website = serializers.URLField()
# Statistics
average_rating = serializers.DecimalField(max_digits=3, decimal_places=2, allow_null=True)
coaster_count = serializers.IntegerField(allow_null=True)
ride_count = serializers.IntegerField(allow_null=True)
# Location (full details)
location = ParkLocationOutputSerializer(allow_null=True)
# Companies
operator = CompanyOutputSerializer()
property_owner = CompanyOutputSerializer(allow_null=True)
# Areas
areas = ParkAreaOutputSerializer(many=True)
# Metadata
created_at = serializers.DateTimeField()
updated_at = serializers.DateTimeField()
class ParkCreateInputSerializer(serializers.Serializer):
"""Input serializer for creating parks."""
name = serializers.CharField(max_length=255)
description = serializers.CharField(allow_blank=True, default="")
status = serializers.ChoiceField(
choices=Park.STATUS_CHOICES,
default="OPERATING"
)
# Optional details
opening_date = serializers.DateField(required=False, allow_null=True)
closing_date = serializers.DateField(required=False, allow_null=True)
operating_season = serializers.CharField(max_length=255, required=False, allow_blank=True)
size_acres = serializers.DecimalField(
max_digits=10,
decimal_places=2,
required=False,
allow_null=True
)
website = serializers.URLField(required=False, allow_blank=True)
# Required operator
operator_id = serializers.IntegerField()
# Optional property owner
property_owner_id = serializers.IntegerField(required=False, allow_null=True)
def validate(self, data):
"""Cross-field validation."""
opening_date = data.get('opening_date')
closing_date = data.get('closing_date')
if opening_date and closing_date and closing_date < opening_date:
raise serializers.ValidationError(
"Closing date cannot be before opening date"
)
return data
class ParkUpdateInputSerializer(serializers.Serializer):
"""Input serializer for updating parks."""
name = serializers.CharField(max_length=255, required=False)
description = serializers.CharField(allow_blank=True, required=False)
status = serializers.ChoiceField(
choices=Park.STATUS_CHOICES,
required=False
)
# Optional details
opening_date = serializers.DateField(required=False, allow_null=True)
closing_date = serializers.DateField(required=False, allow_null=True)
operating_season = serializers.CharField(max_length=255, required=False, allow_blank=True)
size_acres = serializers.DecimalField(
max_digits=10,
decimal_places=2,
required=False,
allow_null=True
)
website = serializers.URLField(required=False, allow_blank=True)
# Companies
operator_id = serializers.IntegerField(required=False)
property_owner_id = serializers.IntegerField(required=False, allow_null=True)
def validate(self, data):
"""Cross-field validation."""
opening_date = data.get('opening_date')
closing_date = data.get('closing_date')
if opening_date and closing_date and closing_date < opening_date:
raise serializers.ValidationError(
"Closing date cannot be before opening date"
)
return data
class ParkFilterInputSerializer(serializers.Serializer):
"""Input serializer for park filtering and search."""
# Search
search = serializers.CharField(required=False, allow_blank=True)
# Status filter
status = serializers.MultipleChoiceField(
choices=Park.STATUS_CHOICES,
required=False
)
# Location filters
country = serializers.CharField(required=False, allow_blank=True)
state = serializers.CharField(required=False, allow_blank=True)
city = serializers.CharField(required=False, allow_blank=True)
# Rating filter
min_rating = serializers.DecimalField(
max_digits=3,
decimal_places=2,
required=False,
min_value=1,
max_value=10
)
# Size filter
min_size_acres = serializers.DecimalField(
max_digits=10,
decimal_places=2,
required=False,
min_value=0
)
max_size_acres = serializers.DecimalField(
max_digits=10,
decimal_places=2,
required=False,
min_value=0
)
# Company filters
operator_id = serializers.IntegerField(required=False)
property_owner_id = serializers.IntegerField(required=False)
# Ordering
ordering = serializers.ChoiceField(
choices=[
'name', '-name',
'opening_date', '-opening_date',
'average_rating', '-average_rating',
'coaster_count', '-coaster_count',
'created_at', '-created_at'
],
required=False,
default='name'
)
class ParkReviewOutputSerializer(serializers.Serializer):
"""Output serializer for park reviews."""
id = serializers.IntegerField()
rating = serializers.IntegerField()
title = serializers.CharField()
content = serializers.CharField()
visit_date = serializers.DateField()
created_at = serializers.DateTimeField()
# User info (limited for privacy)
user = serializers.SerializerMethodField()
def get_user(self, obj):
return {
'username': obj.user.username,
'display_name': obj.user.get_full_name() or obj.user.username
}
class ParkStatsOutputSerializer(serializers.Serializer):
"""Output serializer for park statistics."""
total_parks = serializers.IntegerField()
operating_parks = serializers.IntegerField()
closed_parks = serializers.IntegerField()
under_construction = serializers.IntegerField()
# Averages
average_rating = serializers.DecimalField(max_digits=3, decimal_places=2, allow_null=True)
average_coaster_count = serializers.DecimalField(max_digits=5, decimal_places=2, allow_null=True)
# Top countries
top_countries = serializers.ListField(child=serializers.DictField())
# Recently added
recently_added_count = serializers.IntegerField()

61
parks/api/urls.py Normal file
View File

@@ -0,0 +1,61 @@
"""
URL configuration for Parks API following Django styleguide patterns.
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import (
ParkListApi,
ParkDetailApi,
ParkCreateApi,
ParkUpdateApi,
ParkDeleteApi,
ParkApi
)
app_name = 'parks_api'
# Option 1: Separate ViewSets for each operation (more explicit)
router_separate = DefaultRouter()
router_separate.register(r'list', ParkListApi, basename='park-list')
router_separate.register(r'detail', ParkDetailApi, basename='park-detail')
router_separate.register(r'create', ParkCreateApi, basename='park-create')
router_separate.register(r'update', ParkUpdateApi, basename='park-update')
router_separate.register(r'delete', ParkDeleteApi, basename='park-delete')
# Option 2: Unified ViewSet (more conventional DRF)
router_unified = DefaultRouter()
router_unified.register(r'parks', ParkApi, basename='park')
# Use unified approach for cleaner URLs
urlpatterns = [
path('v1/', include(router_unified.urls)),
]
# Alternative manual URL patterns for more control
urlpatterns_manual = [
# List and create
path('v1/parks/', ParkApi.as_view({
'get': 'list',
'post': 'create'
}), name='park-list'),
# Stats endpoint
path('v1/parks/stats/', ParkApi.as_view({
'get': 'stats'
}), name='park-stats'),
# Detail operations
path('v1/parks/<slug:slug>/', ParkApi.as_view({
'get': 'retrieve',
'put': 'update',
'patch': 'partial_update',
'delete': 'destroy'
}), name='park-detail'),
# Park reviews
path('v1/parks/<slug:slug>/reviews/', ParkApi.as_view({
'get': 'reviews'
}), name='park-reviews'),
]

314
parks/api/views.py Normal file
View File

@@ -0,0 +1,314 @@
"""
Parks API views following Django styleguide patterns.
Uses ClassNameApi naming convention and proper Input/Output serializers.
"""
from typing import Any, Dict
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.viewsets import GenericViewSet
from rest_framework.permissions import IsAuthenticated, IsAuthenticatedOrReadOnly
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework.filters import SearchFilter, OrderingFilter
from core.api.mixins import (
ApiMixin,
CreateApiMixin,
UpdateApiMixin,
ListApiMixin,
RetrieveApiMixin,
DestroyApiMixin
)
from ..selectors import (
park_list_with_stats,
park_detail_optimized,
park_reviews_for_park,
park_statistics
)
from ..services import ParkService
from .serializers import (
ParkListOutputSerializer,
ParkDetailOutputSerializer,
ParkCreateInputSerializer,
ParkUpdateInputSerializer,
ParkFilterInputSerializer,
ParkReviewOutputSerializer,
ParkStatsOutputSerializer
)
class ParkListApi(
ListApiMixin,
GenericViewSet
):
"""
API endpoint for listing parks with filtering and search.
GET /api/v1/parks/
"""
permission_classes = [IsAuthenticatedOrReadOnly]
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
search_fields = ['name', 'description']
ordering_fields = ['name', 'opening_date', 'average_rating', 'coaster_count', 'created_at']
ordering = ['name']
OutputSerializer = ParkListOutputSerializer
FilterSerializer = ParkFilterInputSerializer
def get_queryset(self):
"""Use selector to get optimized queryset."""
# Parse filter parameters
filter_serializer = self.FilterSerializer(data=self.request.query_params)
filter_serializer.is_valid(raise_exception=True)
filters = filter_serializer.validated_data
return park_list_with_stats(filters=filters)
@action(detail=False, methods=['get'])
def stats(self, request: Request) -> Response:
"""
Get park statistics.
GET /api/v1/parks/stats/
"""
stats = park_statistics()
serializer = ParkStatsOutputSerializer(stats)
return self.create_response(
data=serializer.data,
metadata={'cache_duration': 3600} # 1 hour cache hint
)
class ParkDetailApi(
RetrieveApiMixin,
GenericViewSet
):
"""
API endpoint for retrieving individual park details.
GET /api/v1/parks/{id}/
"""
permission_classes = [IsAuthenticatedOrReadOnly]
lookup_field = 'slug'
OutputSerializer = ParkDetailOutputSerializer
def get_object(self):
"""Use selector for optimized detail query."""
slug = self.kwargs.get('slug')
return park_detail_optimized(slug=slug)
@action(detail=True, methods=['get'])
def reviews(self, request: Request, slug: str = None) -> Response:
"""
Get reviews for a specific park.
GET /api/v1/parks/{slug}/reviews/
"""
park = self.get_object()
reviews = park_reviews_for_park(park_id=park.id, limit=50)
serializer = ParkReviewOutputSerializer(reviews, many=True)
return self.create_response(
data=serializer.data,
metadata={
'total_reviews': len(reviews),
'park_name': park.name
}
)
class ParkCreateApi(
CreateApiMixin,
GenericViewSet
):
"""
API endpoint for creating parks.
POST /api/v1/parks/create/
"""
permission_classes = [IsAuthenticated]
InputSerializer = ParkCreateInputSerializer
OutputSerializer = ParkDetailOutputSerializer
def perform_create(self, **validated_data):
"""Create park using service layer."""
return ParkService.create_park(**validated_data)
class ParkUpdateApi(
UpdateApiMixin,
RetrieveApiMixin,
GenericViewSet
):
"""
API endpoint for updating parks.
PUT /api/v1/parks/{slug}/update/
PATCH /api/v1/parks/{slug}/update/
"""
permission_classes = [IsAuthenticated]
lookup_field = 'slug'
InputSerializer = ParkUpdateInputSerializer
OutputSerializer = ParkDetailOutputSerializer
def get_object(self):
"""Use selector for optimized detail query."""
slug = self.kwargs.get('slug')
return park_detail_optimized(slug=slug)
def perform_update(self, instance, **validated_data):
"""Update park using service layer."""
return ParkService.update_park(
park_id=instance.id,
**validated_data
)
class ParkDeleteApi(
DestroyApiMixin,
RetrieveApiMixin,
GenericViewSet
):
"""
API endpoint for deleting parks.
DELETE /api/v1/parks/{slug}/delete/
"""
permission_classes = [IsAuthenticated] # TODO: Add staff/admin permission
lookup_field = 'slug'
def get_object(self):
"""Use selector for optimized detail query."""
slug = self.kwargs.get('slug')
return park_detail_optimized(slug=slug)
def perform_destroy(self, instance):
"""Delete park using service layer."""
ParkService.delete_park(park_id=instance.id)
# Unified API ViewSet (alternative approach)
class ParkApi(
CreateApiMixin,
UpdateApiMixin,
ListApiMixin,
RetrieveApiMixin,
DestroyApiMixin,
GenericViewSet
):
"""
Unified API endpoint for parks with all CRUD operations.
GET /api/v1/parks/ - List parks
POST /api/v1/parks/ - Create park
GET /api/v1/parks/{slug}/ - Get park detail
PUT /api/v1/parks/{slug}/ - Update park
PATCH /api/v1/parks/{slug}/ - Partial update park
DELETE /api/v1/parks/{slug}/ - Delete park
"""
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']
# Serializers for different operations
InputSerializer = ParkCreateInputSerializer # Used for create
UpdateInputSerializer = ParkUpdateInputSerializer # Used for update
OutputSerializer = ParkDetailOutputSerializer # Used for retrieve
ListOutputSerializer = ParkListOutputSerializer # Used for list
FilterSerializer = ParkFilterInputSerializer
def get_queryset(self):
"""Use selector to get optimized queryset."""
if self.action == 'list':
# Parse filter parameters for list view
filter_serializer = self.FilterSerializer(data=self.request.query_params)
filter_serializer.is_valid(raise_exception=True)
filters = filter_serializer.validated_data
return park_list_with_stats(**filters)
# For detail views, this won't be used since we override get_object
return []
def get_object(self):
"""Use selector for optimized detail query."""
slug = self.kwargs.get('slug')
return park_detail_optimized(slug=slug)
def get_output_serializer(self, *args, **kwargs):
"""Return appropriate output serializer based on action."""
if self.action == 'list':
return self.ListOutputSerializer(*args, **kwargs)
return self.OutputSerializer(*args, **kwargs)
def get_input_serializer(self, *args, **kwargs):
"""Return appropriate input serializer based on action."""
if self.action in ['update', 'partial_update']:
return self.UpdateInputSerializer(*args, **kwargs)
return self.InputSerializer(*args, **kwargs)
def perform_create(self, **validated_data):
"""Create park using service layer."""
return ParkService.create_park(**validated_data)
def perform_update(self, instance, **validated_data):
"""Update park using service layer."""
return ParkService.update_park(
park_id=instance.id,
**validated_data
)
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.
GET /api/v1/parks/stats/
"""
stats = park_statistics()
serializer = ParkStatsOutputSerializer(stats)
return self.create_response(
data=serializer.data,
metadata={'cache_duration': 3600}
)
@action(detail=True, methods=['get'])
def reviews(self, request: Request, slug: str = None) -> Response:
"""
Get reviews for a specific park.
GET /api/v1/parks/{slug}/reviews/
"""
park = self.get_object()
reviews = park_reviews_for_park(park_id=park.id, limit=50)
serializer = ParkReviewOutputSerializer(reviews, many=True)
return self.create_response(
data=serializer.data,
metadata={
'total_reviews': len(reviews),
'park_name': park.name
}
)

281
parks/managers.py Normal file
View File

@@ -0,0 +1,281 @@
"""
Custom managers and QuerySets for Parks models.
Optimized queries following Django styleguide patterns.
"""
from typing import Optional, List, Dict, Any, Union
from django.db import models
from django.db.models import Q, F, Count, Avg, Max, Min, Prefetch
from django.contrib.gis.geos import Point
from django.contrib.gis.measure import Distance
from core.managers import (
BaseQuerySet, BaseManager, LocationQuerySet, LocationManager,
ReviewableQuerySet, ReviewableManager, StatusQuerySet, StatusManager
)
class ParkQuerySet(StatusQuerySet, ReviewableQuerySet, LocationQuerySet):
"""Optimized QuerySet for Park model."""
def with_complete_stats(self):
"""Add comprehensive park statistics."""
return self.annotate(
ride_count_calculated=Count('rides', distinct=True),
coaster_count_calculated=Count(
'rides',
filter=Q(rides__category__in=['RC', 'WC']),
distinct=True
),
area_count=Count('areas', distinct=True),
review_count=Count('reviews', filter=Q(reviews__is_published=True), distinct=True),
average_rating_calculated=Avg('reviews__rating', filter=Q(reviews__is_published=True)),
latest_ride_opening=Max('rides__opening_date'),
oldest_ride_opening=Min('rides__opening_date')
)
def optimized_for_list(self):
"""Optimize for park list display."""
return self.select_related(
'operator',
'property_owner'
).prefetch_related(
'location'
).with_complete_stats()
def optimized_for_detail(self):
"""Optimize for park detail display."""
from rides.models import Ride
from .models import ParkReview
return self.select_related(
'operator',
'property_owner'
).prefetch_related(
'location',
'areas',
Prefetch(
'rides',
queryset=Ride.objects.select_related(
'manufacturer', 'designer', 'ride_model', 'park_area'
).order_by('name')
),
Prefetch(
'reviews',
queryset=ParkReview.objects.select_related('user')
.filter(is_published=True)
.order_by('-created_at')[:10]
),
'photos'
)
def by_operator(self, *, operator_id: int):
"""Filter parks by operator."""
return self.filter(operator_id=operator_id)
def by_property_owner(self, *, owner_id: int):
"""Filter parks by property owner."""
return self.filter(property_owner_id=owner_id)
def with_minimum_coasters(self, *, min_coasters: int = 5):
"""Filter parks with minimum number of coasters."""
return self.with_complete_stats().filter(coaster_count_calculated__gte=min_coasters)
def large_parks(self, *, min_acres: float = 100.0):
"""Filter for large parks."""
return self.filter(size_acres__gte=min_acres)
def seasonal_parks(self):
"""Filter for parks with seasonal operation."""
return self.exclude(operating_season__exact='')
def for_map_display(self, *, bounds=None):
"""Optimize for map display with minimal data."""
queryset = self.select_related('operator').prefetch_related('location')
if bounds:
queryset = queryset.within_bounds(
north=bounds.north,
south=bounds.south,
east=bounds.east,
west=bounds.west
)
return queryset.values(
'id', 'name', 'slug', 'status',
'location__latitude', 'location__longitude',
'location__city', 'location__state', 'location__country',
'operator__name'
)
def search_autocomplete(self, *, query: str, limit: int = 10):
"""Optimized search for autocomplete."""
return self.filter(
Q(name__icontains=query) |
Q(location__city__icontains=query) |
Q(location__state__icontains=query)
).select_related('operator', 'location').values(
'id', 'name', 'slug',
'location__city', 'location__state',
'operator__name'
)[:limit]
class ParkManager(StatusManager, ReviewableManager, LocationManager):
"""Custom manager for Park model."""
def get_queryset(self):
return ParkQuerySet(self.model, using=self._db)
def with_complete_stats(self):
return self.get_queryset().with_complete_stats()
def optimized_for_list(self):
return self.get_queryset().optimized_for_list()
def optimized_for_detail(self):
return self.get_queryset().optimized_for_detail()
def by_operator(self, *, operator_id: int):
return self.get_queryset().by_operator(operator_id=operator_id)
def large_parks(self, *, min_acres: float = 100.0):
return self.get_queryset().large_parks(min_acres=min_acres)
def for_map_display(self, *, bounds=None):
return self.get_queryset().for_map_display(bounds=bounds)
class ParkAreaQuerySet(BaseQuerySet):
"""QuerySet for ParkArea model."""
def with_ride_counts(self):
"""Add ride count annotations."""
return self.annotate(
ride_count=Count('rides', distinct=True),
coaster_count=Count(
'rides',
filter=Q(rides__category__in=['RC', 'WC']),
distinct=True
)
)
def optimized_for_list(self):
"""Optimize for area list display."""
return self.select_related('park').with_ride_counts()
def by_park(self, *, park_id: int):
"""Filter areas by park."""
return self.filter(park_id=park_id)
def with_rides(self):
"""Filter areas that have rides."""
return self.filter(rides__isnull=False).distinct()
class ParkAreaManager(BaseManager):
"""Manager for ParkArea model."""
def get_queryset(self):
return ParkAreaQuerySet(self.model, using=self._db)
def with_ride_counts(self):
return self.get_queryset().with_ride_counts()
def by_park(self, *, park_id: int):
return self.get_queryset().by_park(park_id=park_id)
class ParkReviewQuerySet(ReviewableQuerySet):
"""QuerySet for ParkReview model."""
def for_park(self, *, park_id: int):
"""Filter reviews for a specific park."""
return self.filter(park_id=park_id)
def by_user(self, *, user_id: int):
"""Filter reviews by user."""
return self.filter(user_id=user_id)
def by_rating_range(self, *, min_rating: int = 1, max_rating: int = 10):
"""Filter reviews by rating range."""
return self.filter(rating__gte=min_rating, rating__lte=max_rating)
def optimized_for_display(self):
"""Optimize for review display."""
return self.select_related('user', 'park', 'moderated_by')
def recent_reviews(self, *, days: int = 30):
"""Get recent reviews."""
return self.recent(days=days)
def moderation_required(self):
"""Filter reviews requiring moderation."""
return self.filter(
Q(is_published=False) |
Q(moderated_at__isnull=True)
)
class ParkReviewManager(BaseManager):
"""Manager for ParkReview model."""
def get_queryset(self):
return ParkReviewQuerySet(self.model, using=self._db)
def for_park(self, *, park_id: int):
return self.get_queryset().for_park(park_id=park_id)
def by_rating_range(self, *, min_rating: int = 1, max_rating: int = 10):
return self.get_queryset().by_rating_range(min_rating=min_rating, max_rating=max_rating)
def moderation_required(self):
return self.get_queryset().moderation_required()
class CompanyQuerySet(BaseQuerySet):
"""QuerySet for Company model."""
def operators(self):
"""Filter for companies that operate parks."""
return self.filter(roles__contains=['OPERATOR'])
def property_owners(self):
"""Filter for companies that own park properties."""
return self.filter(roles__contains=['PROPERTY_OWNER'])
def manufacturers(self):
"""Filter for ride manufacturers."""
return self.filter(roles__contains=['MANUFACTURER'])
def with_park_counts(self):
"""Add park count annotations."""
return self.annotate(
operated_parks_count=Count('operated_parks', distinct=True),
owned_parks_count=Count('owned_parks', distinct=True),
total_parks_involvement=Count('operated_parks', distinct=True) + Count('owned_parks', distinct=True)
)
def major_operators(self, *, min_parks: int = 5):
"""Filter for major park operators."""
return self.operators().with_park_counts().filter(operated_parks_count__gte=min_parks)
def optimized_for_list(self):
"""Optimize for company list display."""
return self.with_park_counts()
class CompanyManager(BaseManager):
"""Manager for Company model."""
def get_queryset(self):
return CompanyQuerySet(self.model, using=self._db)
def operators(self):
return self.get_queryset().operators()
def manufacturers(self):
return self.get_queryset().manufacturers()
def major_operators(self, *, min_parks: int = 5):
return self.get_queryset().major_operators(min_parks=min_parks)

122
parks/migrations/0003_add_business_constraints.py Normal file
View File

@@ -0,0 +1,122 @@
# Generated by Django 5.2.5 on 2025-08-16 17:42
import django.db.models.functions.datetime
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("parks", "0002_alter_parkarea_unique_together"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.AddConstraint(
model_name="park",
constraint=models.CheckConstraint(
condition=models.Q(
("closing_date__isnull", True),
("opening_date__isnull", True),
("closing_date__gte", models.F("opening_date")),
_connector="OR",
),
name="park_closing_after_opening",
violation_error_message="Closing date must be after opening date",
),
),
migrations.AddConstraint(
model_name="park",
constraint=models.CheckConstraint(
condition=models.Q(
("size_acres__isnull", True), ("size_acres__gt", 0), _connector="OR"
),
name="park_size_positive",
violation_error_message="Park size must be positive",
),
),
migrations.AddConstraint(
model_name="park",
constraint=models.CheckConstraint(
condition=models.Q(
("average_rating__isnull", True),
models.Q(("average_rating__gte", 1), ("average_rating__lte", 10)),
_connector="OR",
),
name="park_rating_range",
violation_error_message="Average rating must be between 1 and 10",
),
),
migrations.AddConstraint(
model_name="park",
constraint=models.CheckConstraint(
condition=models.Q(
("ride_count__isnull", True),
("ride_count__gte", 0),
_connector="OR",
),
name="park_ride_count_non_negative",
violation_error_message="Ride count must be non-negative",
),
),
migrations.AddConstraint(
model_name="park",
constraint=models.CheckConstraint(
condition=models.Q(
("coaster_count__isnull", True),
("coaster_count__gte", 0),
_connector="OR",
),
name="park_coaster_count_non_negative",
violation_error_message="Coaster count must be non-negative",
),
),
migrations.AddConstraint(
model_name="park",
constraint=models.CheckConstraint(
condition=models.Q(
("coaster_count__isnull", True),
("ride_count__isnull", True),
("coaster_count__lte", models.F("ride_count")),
_connector="OR",
),
name="park_coaster_count_lte_ride_count",
violation_error_message="Coaster count cannot exceed total ride count",
),
),
migrations.AddConstraint(
model_name="parkreview",
constraint=models.CheckConstraint(
condition=models.Q(("rating__gte", 1), ("rating__lte", 10)),
name="park_review_rating_range",
violation_error_message="Rating must be between 1 and 10",
),
),
migrations.AddConstraint(
model_name="parkreview",
constraint=models.CheckConstraint(
condition=models.Q(
("visit_date__lte", django.db.models.functions.datetime.Now())
),
name="park_review_visit_date_not_future",
violation_error_message="Visit date cannot be in the future",
),
),
migrations.AddConstraint(
model_name="parkreview",
constraint=models.CheckConstraint(
condition=models.Q(
models.Q(
("moderated_at__isnull", True), ("moderated_by__isnull", True)
),
models.Q(
("moderated_at__isnull", False), ("moderated_by__isnull", False)
),
_connector="OR",
),
name="park_review_moderation_consistency",
violation_error_message="Moderated reviews must have both moderator and moderation timestamp",
),
),
]

104
parks/migrations/0004_fix_pghistory_triggers.py Normal file
View File

@@ -0,0 +1,104 @@
# Generated by Django 5.2.5 on 2025-08-16 17:46
import django.contrib.postgres.fields
import django.db.models.deletion
import pgtrigger.compiler
import pgtrigger.migrations
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("parks", "0003_add_business_constraints"),
("pghistory", "0007_auto_20250421_0444"),
]
operations = [
migrations.CreateModel(
name="CompanyEvent",
fields=[
("pgh_id", models.AutoField(primary_key=True, serialize=False)),
("pgh_created_at", models.DateTimeField(auto_now_add=True)),
("pgh_label", models.TextField(help_text="The event label.")),
("id", models.BigIntegerField()),
("created_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
("name", models.CharField(max_length=255)),
("slug", models.SlugField(db_index=False, max_length=255)),
(
"roles",
django.contrib.postgres.fields.ArrayField(
base_field=models.CharField(
choices=[
("OPERATOR", "Park Operator"),
("PROPERTY_OWNER", "Property Owner"),
],
max_length=20,
),
blank=True,
default=list,
size=None,
),
),
("description", models.TextField(blank=True)),
("website", models.URLField(blank=True)),
("founded_year", models.PositiveIntegerField(blank=True, null=True)),
("parks_count", models.IntegerField(default=0)),
("rides_count", models.IntegerField(default=0)),
],
options={
"abstract": False,
},
),
pgtrigger.migrations.AddTrigger(
model_name="company",
trigger=pgtrigger.compiler.Trigger(
name="insert_insert",
sql=pgtrigger.compiler.UpsertTriggerSql(
func='INSERT INTO "parks_companyevent" ("created_at", "description", "founded_year", "id", "name", "parks_count", "pgh_context_id", "pgh_created_at", "pgh_label", "pgh_obj_id", "rides_count", "roles", "slug", "updated_at", "website") VALUES (NEW."created_at", NEW."description", NEW."founded_year", NEW."id", NEW."name", NEW."parks_count", _pgh_attach_context(), NOW(), \'insert\', NEW."id", NEW."rides_count", NEW."roles", NEW."slug", NEW."updated_at", NEW."website"); RETURN NULL;',
hash="[AWS-SECRET-REMOVED]",
operation="INSERT",
pgid="pgtrigger_insert_insert_35b57",
table="parks_company",
when="AFTER",
),
),
),
pgtrigger.migrations.AddTrigger(
model_name="company",
trigger=pgtrigger.compiler.Trigger(
name="update_update",
sql=pgtrigger.compiler.UpsertTriggerSql(
condition="WHEN (OLD.* IS DISTINCT FROM NEW.*)",
func='INSERT INTO "parks_companyevent" ("created_at", "description", "founded_year", "id", "name", "parks_count", "pgh_context_id", "pgh_created_at", "pgh_label", "pgh_obj_id", "rides_count", "roles", "slug", "updated_at", "website") VALUES (NEW."created_at", NEW."description", NEW."founded_year", NEW."id", NEW."name", NEW."parks_count", _pgh_attach_context(), NOW(), \'update\', NEW."id", NEW."rides_count", NEW."roles", NEW."slug", NEW."updated_at", NEW."website"); RETURN NULL;',
hash="[AWS-SECRET-REMOVED]",
operation="UPDATE",
pgid="pgtrigger_update_update_d3286",
table="parks_company",
when="AFTER",
),
),
),
migrations.AddField(
model_name="companyevent",
name="pgh_context",
field=models.ForeignKey(
db_constraint=False,
null=True,
on_delete=django.db.models.deletion.DO_NOTHING,
related_name="+",
to="pghistory.context",
),
),
migrations.AddField(
model_name="companyevent",
name="pgh_obj",
field=models.ForeignKey(
db_constraint=False,
on_delete=django.db.models.deletion.DO_NOTHING,
related_name="events",
to="parks.company",
),
),
]

5
parks/models/areas.py
View File

@@ -9,6 +9,11 @@ from .parks import Park
@pghistory.track()
class ParkArea(TrackedModel):
# Import managers
from ..managers import ParkAreaManager
objects = ParkAreaManager()
id: int # Type hint for Django's automatic id field
park = models.ForeignKey(Park, on_delete=models.CASCADE, related_name="areas")
name = models.CharField(max_length=255)

7
parks/models/companies.py
View File

@@ -1,8 +1,15 @@
from django.contrib.postgres.fields import ArrayField
from django.db import models
from core.models import TrackedModel
import pghistory
@pghistory.track()
class Company(TrackedModel):
# Import managers
from ..managers import CompanyManager
objects = CompanyManager()
class CompanyRole(models.TextChoices):
OPERATOR = 'OPERATOR', 'Park Operator'
PROPERTY_OWNER = 'PROPERTY_OWNER', 'Property Owner'

42
parks/models/parks.py
View File

@@ -17,6 +17,11 @@ if TYPE_CHECKING:
@pghistory.track()
class Park(TrackedModel):
# Import managers
from ..managers import ParkManager
objects = ParkManager()
id: int # Type hint for Django's automatic id field
STATUS_CHOICES = [
("OPERATING", "Operating"),
@@ -81,6 +86,43 @@ class Park(TrackedModel):
class Meta:
ordering = ["name"]
constraints = [
# Business rule: Closing date must be after opening date
models.CheckConstraint(
name="park_closing_after_opening",
check=models.Q(closing_date__isnull=True) | models.Q(opening_date__isnull=True) | models.Q(closing_date__gte=models.F("opening_date")),
violation_error_message="Closing date must be after opening date"
),
# Business rule: Size must be positive
models.CheckConstraint(
name="park_size_positive",
check=models.Q(size_acres__isnull=True) | models.Q(size_acres__gt=0),
violation_error_message="Park size must be positive"
),
# Business rule: Rating must be between 1 and 10
models.CheckConstraint(
name="park_rating_range",
check=models.Q(average_rating__isnull=True) | (models.Q(average_rating__gte=1) & models.Q(average_rating__lte=10)),
violation_error_message="Average rating must be between 1 and 10"
),
# Business rule: Counts must be non-negative
models.CheckConstraint(
name="park_ride_count_non_negative",
check=models.Q(ride_count__isnull=True) | models.Q(ride_count__gte=0),
violation_error_message="Ride count must be non-negative"
),
models.CheckConstraint(
name="park_coaster_count_non_negative",
check=models.Q(coaster_count__isnull=True) | models.Q(coaster_count__gte=0),
violation_error_message="Coaster count must be non-negative"
),
# Business rule: Coaster count cannot exceed ride count
models.CheckConstraint(
name="park_coaster_count_lte_ride_count",
check=models.Q(coaster_count__isnull=True) | models.Q(ride_count__isnull=True) | models.Q(coaster_count__lte=models.F("ride_count")),
violation_error_message="Coaster count cannot exceed total ride count"
),
]
def __str__(self) -> str:
return self.name

27
parks/models/reviews.py
View File

@@ -1,10 +1,16 @@
from django.db import models
from django.db.models import functions
from django.core.validators import MinValueValidator, MaxValueValidator
from core.history import TrackedModel
import pghistory
@pghistory.track()
class ParkReview(TrackedModel):
# Import managers
from ..managers import ParkReviewManager
objects = ParkReviewManager()
"""
A review of a park.
"""
@@ -44,6 +50,27 @@ class ParkReview(TrackedModel):
class Meta:
ordering = ['-created_at']
unique_together = ['park', 'user']
constraints = [
# Business rule: Rating must be between 1 and 10 (database level enforcement)
models.CheckConstraint(
name="park_review_rating_range",
check=models.Q(rating__gte=1) & models.Q(rating__lte=10),
violation_error_message="Rating must be between 1 and 10"
),
# Business rule: Visit date cannot be in the future
models.CheckConstraint(
name="park_review_visit_date_not_future",
check=models.Q(visit_date__lte=functions.Now()),
violation_error_message="Visit date cannot be in the future"
),
# Business rule: If moderated, must have moderator and timestamp
models.CheckConstraint(
name="park_review_moderation_consistency",
check=models.Q(moderated_by__isnull=True, moderated_at__isnull=True) |
models.Q(moderated_by__isnull=False, moderated_at__isnull=False),
violation_error_message="Moderated reviews must have both moderator and moderation timestamp"
),
]
def __str__(self):
return f"Review of {self.park.name} by {self.user.username}"

244
parks/selectors.py Normal file
View File

@@ -0,0 +1,244 @@
"""
Selectors for park-related data retrieval.
Following Django styleguide pattern for separating data access from business logic.
"""
from typing import Optional, Dict, Any, List
from django.db.models import QuerySet, Q, F, Count, Avg, Prefetch
from django.contrib.gis.geos import Point
from django.contrib.gis.measure import Distance
from .models import Park, ParkArea, ParkReview
from rides.models import Ride
def park_list_with_stats(*, filters: Optional[Dict[str, Any]] = None) -> QuerySet[Park]:
"""
Get parks optimized for list display with basic stats.
Args:
filters: Optional dictionary of filter parameters
Returns:
QuerySet of parks with optimized queries
"""
queryset = Park.objects.select_related(
'operator',
'property_owner'
).prefetch_related(
'location'
).annotate(
ride_count_calculated=Count('rides', distinct=True),
coaster_count_calculated=Count(
'rides',
filter=Q(rides__category__in=['RC', 'WC']),
distinct=True
),
average_rating_calculated=Avg('reviews__rating')
)
if filters:
if 'status' in filters:
queryset = queryset.filter(status=filters['status'])
if 'operator' in filters:
queryset = queryset.filter(operator=filters['operator'])
if 'country' in filters:
queryset = queryset.filter(location__country=filters['country'])
if 'search' in filters:
search_term = filters['search']
queryset = queryset.filter(
Q(name__icontains=search_term) |
Q(description__icontains=search_term)
)
return queryset.order_by('name')
def park_detail_optimized(*, slug: str) -> Park:
"""
Get a single park with all related data optimized for detail view.
Args:
slug: Park slug identifier
Returns:
Park instance with optimized prefetches
Raises:
Park.DoesNotExist: If park with slug doesn't exist
"""
return Park.objects.select_related(
'operator',
'property_owner'
).prefetch_related(
'location',
'areas',
Prefetch(
'rides',
queryset=Ride.objects.select_related('manufacturer', 'designer', 'ride_model')
),
Prefetch(
'reviews',
queryset=ParkReview.objects.select_related('user').filter(is_published=True)
),
'photos'
).get(slug=slug)
def parks_near_location(
*,
point: Point,
distance_km: float = 50,
limit: int = 10
) -> QuerySet[Park]:
"""
Get parks near a specific geographic location.
Args:
point: Geographic point (longitude, latitude)
distance_km: Maximum distance in kilometers
limit: Maximum number of results
Returns:
QuerySet of nearby parks ordered by distance
"""
return Park.objects.filter(
location__coordinates__distance_lte=(point, Distance(km=distance_km))
).select_related(
'operator'
).prefetch_related(
'location'
).distance(point).order_by('distance')[:limit]
def park_statistics() -> Dict[str, Any]:
"""
Get overall park statistics for dashboard/analytics.
Returns:
Dictionary containing park statistics
"""
total_parks = Park.objects.count()
operating_parks = Park.objects.filter(status='OPERATING').count()
total_rides = Ride.objects.count()
total_coasters = Ride.objects.filter(category__in=['RC', 'WC']).count()
return {
'total_parks': total_parks,
'operating_parks': operating_parks,
'closed_parks': total_parks - operating_parks,
'total_rides': total_rides,
'total_coasters': total_coasters,
'average_rides_per_park': total_rides / total_parks if total_parks > 0 else 0
}
def parks_by_operator(*, operator_id: int) -> QuerySet[Park]:
"""
Get all parks operated by a specific company.
Args:
operator_id: Company ID of the operator
Returns:
QuerySet of parks operated by the company
"""
return Park.objects.filter(
operator_id=operator_id
).select_related(
'operator'
).prefetch_related(
'location'
).annotate(
ride_count_calculated=Count('rides')
).order_by('name')
def parks_with_recent_reviews(*, days: int = 30) -> QuerySet[Park]:
"""
Get parks that have received reviews in the last N days.
Args:
days: Number of days to look back for reviews
Returns:
QuerySet of parks with recent reviews
"""
from django.utils import timezone
from datetime import timedelta
cutoff_date = timezone.now() - timedelta(days=days)
return Park.objects.filter(
reviews__created_at__gte=cutoff_date,
reviews__is_published=True
).select_related(
'operator'
).prefetch_related(
'location'
).annotate(
recent_review_count=Count('reviews', filter=Q(reviews__created_at__gte=cutoff_date))
).order_by('-recent_review_count').distinct()
def park_search_autocomplete(*, query: str, limit: int = 10) -> QuerySet[Park]:
"""
Get parks matching a search query for autocomplete functionality.
Args:
query: Search string
limit: Maximum number of results
Returns:
QuerySet of matching parks for autocomplete
"""
return Park.objects.filter(
Q(name__icontains=query) |
Q(location__city__icontains=query) |
Q(location__region__icontains=query)
).select_related(
'operator'
).prefetch_related(
'location'
).order_by('name')[:limit]
def park_areas_for_park(*, park_slug: str) -> QuerySet[ParkArea]:
"""
Get all areas for a specific park.
Args:
park_slug: Slug of the park
Returns:
QuerySet of park areas with related data
"""
return ParkArea.objects.filter(
park__slug=park_slug
).select_related(
'park'
).prefetch_related(
'rides'
).annotate(
ride_count=Count('rides')
).order_by('name')
def park_reviews_for_park(*, park_id: int, limit: int = 20) -> QuerySet[ParkReview]:
"""
Get reviews for a specific park.
Args:
park_id: Park ID
limit: Maximum number of reviews to return
Returns:
QuerySet of park reviews
"""
return ParkReview.objects.filter(
park_id=park_id,
is_published=True
).select_related(
'user',
'park'
).order_by('-created_at')[:limit]

333
parks/services.py Normal file
View File

@@ -0,0 +1,333 @@
"""
Services for park-related business logic.
Following Django styleguide pattern for business logic encapsulation.
"""
from typing import Optional, Dict, Any, Tuple
from django.db import transaction
from django.db.models import Q
from django.core.exceptions import ValidationError
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AbstractBaseUser
from .models import Park, ParkArea
from location.models import Location
# Use AbstractBaseUser for type hinting
UserType = AbstractBaseUser
User = get_user_model()
class ParkService:
"""Service for managing park operations."""
@staticmethod
def create_park(
*,
name: str,
description: str = "",
status: str = "OPERATING",
operator_id: Optional[int] = None,
property_owner_id: Optional[int] = None,
opening_date: Optional[str] = None,
closing_date: Optional[str] = None,
operating_season: str = "",
size_acres: Optional[float] = None,
website: str = "",
location_data: Optional[Dict[str, Any]] = None,
created_by: Optional[UserType] = None
) -> Park:
"""
Create a new park with validation and location handling.
Args:
name: Park name
description: Park description
status: Operating status
operator_id: ID of operating company
property_owner_id: ID of property owner company
opening_date: Opening date
closing_date: Closing date
operating_season: Operating season description
size_acres: Park size in acres
website: Park website URL
location_data: Dictionary containing location information
created_by: User creating the park
Returns:
Created Park instance
Raises:
ValidationError: If park data is invalid
"""
with transaction.atomic():
# Create park instance
park = Park(
name=name,
description=description,
status=status,
opening_date=opening_date,
closing_date=closing_date,
operating_season=operating_season,
size_acres=size_acres,
website=website
)
# Set foreign key relationships if provided
if operator_id:
from .models import Company
park.operator = Company.objects.get(id=operator_id)
if property_owner_id:
from .models import Company
park.property_owner = Company.objects.get(id=property_owner_id)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
park.full_clean()
park.save()
# Handle location if provided
if location_data:
LocationService.create_park_location(
park=park,
**location_data
)
return park
@staticmethod
def update_park(
*,
park_id: int,
updates: Dict[str, Any],
updated_by: Optional[UserType] = None
) -> Park:
"""
Update an existing park with validation.
Args:
park_id: ID of park to update
updates: Dictionary of field updates
updated_by: User performing the update
Returns:
Updated Park instance
Raises:
Park.DoesNotExist: If park doesn't exist
ValidationError: If update data is invalid
"""
with transaction.atomic():
park = Park.objects.select_for_update().get(id=park_id)
# Apply updates
for field, value in updates.items():
if hasattr(park, field):
setattr(park, field, value)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
park.full_clean()
park.save()
return park
@staticmethod
def delete_park(*, park_id: int, deleted_by: Optional[UserType] = None) -> bool:
"""
Soft delete a park by setting status to DEMOLISHED.
Args:
park_id: ID of park to delete
deleted_by: User performing the deletion
Returns:
True if successfully deleted
Raises:
Park.DoesNotExist: If park doesn't exist
"""
with transaction.atomic():
park = Park.objects.select_for_update().get(id=park_id)
park.status = 'DEMOLISHED'
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
park.full_clean()
park.save()
return True
@staticmethod
def create_park_area(
*,
park_id: int,
name: str,
description: str = "",
created_by: Optional[UserType] = None
) -> ParkArea:
"""
Create a new area within a park.
Args:
park_id: ID of the parent park
name: Area name
description: Area description
created_by: User creating the area
Returns:
Created ParkArea instance
Raises:
Park.DoesNotExist: If park doesn't exist
ValidationError: If area data is invalid
"""
park = Park.objects.get(id=park_id)
area = ParkArea(
park=park,
name=name,
description=description
)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
area.full_clean()
area.save()
return area
@staticmethod
def update_park_statistics(*, park_id: int) -> Park:
"""
Recalculate and update park statistics (ride counts, ratings).
Args:
park_id: ID of park to update statistics for
Returns:
Updated Park instance with fresh statistics
"""
from rides.models import Ride
from .models import ParkReview
from django.db.models import Count, Avg
with transaction.atomic():
park = Park.objects.select_for_update().get(id=park_id)
# Calculate ride counts
ride_stats = Ride.objects.filter(park=park).aggregate(
total_rides=Count('id'),
coaster_count=Count('id', filter=Q(category__in=['RC', 'WC']))
)
# Calculate average rating
avg_rating = ParkReview.objects.filter(
park=park,
is_published=True
).aggregate(avg_rating=Avg('rating'))['avg_rating']
# Update park fields
park.ride_count = ride_stats['total_rides'] or 0
park.coaster_count = ride_stats['coaster_count'] or 0
park.average_rating = avg_rating
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
park.full_clean()
park.save()
return park
class LocationService:
"""Service for managing location operations."""
@staticmethod
def create_park_location(
*,
park: Park,
latitude: Optional[float] = None,
longitude: Optional[float] = None,
street_address: str = "",
city: str = "",
state: str = "",
country: str = "",
postal_code: str = ""
) -> Location:
"""
Create a location for a park.
Args:
park: Park instance
latitude: Latitude coordinate
longitude: Longitude coordinate
street_address: Street address
city: City name
state: State/region name
country: Country name
postal_code: Postal/ZIP code
Returns:
Created Location instance
Raises:
ValidationError: If location data is invalid
"""
location = Location(
content_object=park,
name=park.name,
location_type='park',
latitude=latitude,
longitude=longitude,
street_address=street_address,
city=city,
state=state,
country=country,
postal_code=postal_code
)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
location.full_clean()
location.save()
return location
@staticmethod
def update_park_location(
*,
park_id: int,
location_updates: Dict[str, Any]
) -> Location:
"""
Update location information for a park.
Args:
park_id: ID of the park
location_updates: Dictionary of location field updates
Returns:
Updated Location instance
Raises:
Location.DoesNotExist: If location doesn't exist
ValidationError: If location data is invalid
"""
with transaction.atomic():
park = Park.objects.get(id=park_id)
try:
location = park.location
except Location.DoesNotExist:
# Create location if it doesn't exist
return LocationService.create_park_location(
park=park,
**location_updates
)
# Apply updates
for field, value in location_updates.items():
if hasattr(location, field):
setattr(location, field, value)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
location.full_clean()
location.save()
return location

3
parks/services/__init__.py
View File

@@ -1,3 +1,4 @@
from .roadtrip import RoadTripService
from .park_management import ParkService, LocationService
__all__ = ['RoadTripService']
__all__ = ['RoadTripService', 'ParkService', 'LocationService']

330
parks/services/park_management.py Normal file
View File

@@ -0,0 +1,330 @@
"""
Services for park-related business logic.
Following Django styleguide pattern for business logic encapsulation.
"""
from typing import Optional, Dict, Any, TYPE_CHECKING
from django.db import transaction
from django.db.models import Q
from django.core.exceptions import ValidationError
if TYPE_CHECKING:
from django.contrib.auth.models import AbstractUser
from ..models import Park, ParkArea
from location.models import Location
class ParkService:
"""Service for managing park operations."""
@staticmethod
def create_park(
*,
name: str,
description: str = "",
status: str = "OPERATING",
operator_id: Optional[int] = None,
property_owner_id: Optional[int] = None,
opening_date: Optional[str] = None,
closing_date: Optional[str] = None,
operating_season: str = "",
size_acres: Optional[float] = None,
website: str = "",
location_data: Optional[Dict[str, Any]] = None,
created_by: Optional['AbstractUser'] = None
) -> Park:
"""
Create a new park with validation and location handling.
Args:
name: Park name
description: Park description
status: Operating status
operator_id: ID of operating company
property_owner_id: ID of property owner company
opening_date: Opening date
closing_date: Closing date
operating_season: Operating season description
size_acres: Park size in acres
website: Park website URL
location_data: Dictionary containing location information
created_by: User creating the park
Returns:
Created Park instance
Raises:
ValidationError: If park data is invalid
"""
with transaction.atomic():
# Create park instance
park = Park(
name=name,
description=description,
status=status,
opening_date=opening_date,
closing_date=closing_date,
operating_season=operating_season,
size_acres=size_acres,
website=website
)
# Set foreign key relationships if provided
if operator_id:
from parks.models import Company
park.operator = Company.objects.get(id=operator_id)
if property_owner_id:
from parks.models import Company
park.property_owner = Company.objects.get(id=property_owner_id)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
park.full_clean()
park.save()
# Handle location if provided
if location_data:
LocationService.create_park_location(
park=park,
**location_data
)
return park
@staticmethod
def update_park(
*,
park_id: int,
updates: Dict[str, Any],
updated_by: Optional['AbstractUser'] = None
) -> Park:
"""
Update an existing park with validation.
Args:
park_id: ID of park to update
updates: Dictionary of field updates
updated_by: User performing the update
Returns:
Updated Park instance
Raises:
Park.DoesNotExist: If park doesn't exist
ValidationError: If update data is invalid
"""
with transaction.atomic():
park = Park.objects.select_for_update().get(id=park_id)
# Apply updates
for field, value in updates.items():
if hasattr(park, field):
setattr(park, field, value)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
park.full_clean()
park.save()
return park
@staticmethod
def delete_park(*, park_id: int, deleted_by: Optional['AbstractUser'] = None) -> bool:
"""
Soft delete a park by setting status to DEMOLISHED.
Args:
park_id: ID of park to delete
deleted_by: User performing the deletion
Returns:
True if successfully deleted
Raises:
Park.DoesNotExist: If park doesn't exist
"""
with transaction.atomic():
park = Park.objects.select_for_update().get(id=park_id)
park.status = 'DEMOLISHED'
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
park.full_clean()
park.save()
return True
@staticmethod
def create_park_area(
*,
park_id: int,
name: str,
description: str = "",
created_by: Optional['AbstractUser'] = None
) -> ParkArea:
"""
Create a new area within a park.
Args:
park_id: ID of the parent park
name: Area name
description: Area description
created_by: User creating the area
Returns:
Created ParkArea instance
Raises:
Park.DoesNotExist: If park doesn't exist
ValidationError: If area data is invalid
"""
park = Park.objects.get(id=park_id)
area = ParkArea(
park=park,
name=name,
description=description
)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
area.full_clean()
area.save()
return area
@staticmethod
def update_park_statistics(*, park_id: int) -> Park:
"""
Recalculate and update park statistics (ride counts, ratings).
Args:
park_id: ID of park to update statistics for
Returns:
Updated Park instance with fresh statistics
"""
from rides.models import Ride
from parks.models import ParkReview
from django.db.models import Count, Avg
with transaction.atomic():
park = Park.objects.select_for_update().get(id=park_id)
# Calculate ride counts
ride_stats = Ride.objects.filter(park=park).aggregate(
total_rides=Count('id'),
coaster_count=Count('id', filter=Q(category__in=['RC', 'WC']))
)
# Calculate average rating
avg_rating = ParkReview.objects.filter(
park=park,
is_published=True
).aggregate(avg_rating=Avg('rating'))['avg_rating']
# Update park fields
park.ride_count = ride_stats['total_rides'] or 0
park.coaster_count = ride_stats['coaster_count'] or 0
park.average_rating = avg_rating
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
park.full_clean()
park.save()
return park
class LocationService:
"""Service for managing location operations."""
@staticmethod
def create_park_location(
*,
park: Park,
latitude: Optional[float] = None,
longitude: Optional[float] = None,
street_address: str = "",
city: str = "",
state: str = "",
country: str = "",
postal_code: str = ""
) -> Location:
"""
Create a location for a park.
Args:
park: Park instance
latitude: Latitude coordinate
longitude: Longitude coordinate
street_address: Street address
city: City name
state: State/region name
country: Country name
postal_code: Postal/ZIP code
Returns:
Created Location instance
Raises:
ValidationError: If location data is invalid
"""
location = Location(
content_object=park,
name=park.name,
location_type='park',
latitude=latitude,
longitude=longitude,
street_address=street_address,
city=city,
state=state,
country=country,
postal_code=postal_code
)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
location.full_clean()
location.save()
return location
@staticmethod
def update_park_location(
*,
park_id: int,
location_updates: Dict[str, Any]
) -> Location:
"""
Update location information for a park.
Args:
park_id: ID of the park
location_updates: Dictionary of location field updates
Returns:
Updated Location instance
Raises:
Location.DoesNotExist: If location doesn't exist
ValidationError: If location data is invalid
"""
with transaction.atomic():
park = Park.objects.get(id=park_id)
try:
location = park.location
except Location.DoesNotExist:
# Create location if it doesn't exist
return LocationService.create_park_location(
park=park,
**location_updates
)
# Apply updates
for field, value in location_updates.items():
if hasattr(location, field):
setattr(location, field, value)
# CRITICAL STYLEGUIDE FIX: Call full_clean before save
location.full_clean()
location.save()
return location

2
parks/views.py
View File

@@ -4,7 +4,7 @@ from .models.location import ParkLocation
from media.models import Photo
from moderation.models import EditSubmission
from moderation.mixins import EditSubmissionMixin, PhotoSubmissionMixin, HistoryMixin
from core.views import SlugRedirectMixin
from core.views.views import SlugRedirectMixin
from .filters import ParkFilter
from .forms import ParkForm
from .models import Park, ParkArea, ParkReview as Review