mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2025-12-20 14:31:08 -05:00
c26414ff74
- 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.
300 lines
8.8 KiB
Python
300 lines
8.8 KiB
Python
"""
|
|
Selectors for core functionality including map services and analytics.
|
|
Following Django styleguide pattern for separating data access from business logic.
|
|
"""
|
|
|
|
from typing import Optional, Dict, Any, List, Union
|
|
from django.db.models import QuerySet, Q, F, Count, Avg
|
|
from django.contrib.gis.geos import Point, Polygon
|
|
from django.contrib.gis.measure import Distance
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
|
|
from .analytics import PageView
|
|
from parks.models import Park
|
|
from rides.models import Ride
|
|
|
|
|
|
def unified_locations_for_map(
|
|
*,
|
|
bounds: Optional[Polygon] = None,
|
|
location_types: Optional[List[str]] = None,
|
|
filters: Optional[Dict[str, Any]] = None
|
|
) -> Dict[str, QuerySet]:
|
|
"""
|
|
Get unified location data for map display across all location types.
|
|
|
|
Args:
|
|
bounds: Geographic boundary polygon
|
|
location_types: List of location types to include ('park', 'ride')
|
|
filters: Additional filter parameters
|
|
|
|
Returns:
|
|
Dictionary containing querysets for each location type
|
|
"""
|
|
results = {}
|
|
|
|
# Default to all location types if none specified
|
|
if not location_types:
|
|
location_types = ['park', 'ride']
|
|
|
|
# Parks
|
|
if 'park' in location_types:
|
|
park_queryset = Park.objects.select_related(
|
|
'operator'
|
|
).prefetch_related(
|
|
'location'
|
|
).annotate(
|
|
ride_count_calculated=Count('rides')
|
|
)
|
|
|
|
if bounds:
|
|
park_queryset = park_queryset.filter(
|
|
location__coordinates__within=bounds
|
|
)
|
|
|
|
if filters:
|
|
if 'status' in filters:
|
|
park_queryset = park_queryset.filter(status=filters['status'])
|
|
if 'operator' in filters:
|
|
park_queryset = park_queryset.filter(operator=filters['operator'])
|
|
|
|
results['parks'] = park_queryset.order_by('name')
|
|
|
|
# Rides
|
|
if 'ride' in location_types:
|
|
ride_queryset = Ride.objects.select_related(
|
|
'park',
|
|
'manufacturer'
|
|
).prefetch_related(
|
|
'park__location',
|
|
'location'
|
|
)
|
|
|
|
if bounds:
|
|
ride_queryset = ride_queryset.filter(
|
|
Q(location__coordinates__within=bounds) |
|
|
Q(park__location__coordinates__within=bounds)
|
|
)
|
|
|
|
if filters:
|
|
if 'category' in filters:
|
|
ride_queryset = ride_queryset.filter(category=filters['category'])
|
|
if 'manufacturer' in filters:
|
|
ride_queryset = ride_queryset.filter(manufacturer=filters['manufacturer'])
|
|
if 'park' in filters:
|
|
ride_queryset = ride_queryset.filter(park=filters['park'])
|
|
|
|
results['rides'] = ride_queryset.order_by('park__name', 'name')
|
|
|
|
return results
|
|
|
|
|
|
def locations_near_point(
|
|
*,
|
|
point: Point,
|
|
distance_km: float = 50,
|
|
location_types: Optional[List[str]] = None,
|
|
limit: int = 20
|
|
) -> Dict[str, QuerySet]:
|
|
"""
|
|
Get locations near a specific geographic point across all types.
|
|
|
|
Args:
|
|
point: Geographic point (longitude, latitude)
|
|
distance_km: Maximum distance in kilometers
|
|
location_types: List of location types to include
|
|
limit: Maximum number of results per type
|
|
|
|
Returns:
|
|
Dictionary containing nearby locations by type
|
|
"""
|
|
results = {}
|
|
|
|
if not location_types:
|
|
location_types = ['park', 'ride']
|
|
|
|
# Parks near point
|
|
if 'park' in location_types:
|
|
results['parks'] = Park.objects.filter(
|
|
location__coordinates__distance_lte=(point, Distance(km=distance_km))
|
|
).select_related(
|
|
'operator'
|
|
).prefetch_related(
|
|
'location'
|
|
).distance(point).order_by('distance')[:limit]
|
|
|
|
# Rides near point
|
|
if 'ride' in location_types:
|
|
results['rides'] = Ride.objects.filter(
|
|
Q(location__coordinates__distance_lte=(point, Distance(km=distance_km))) |
|
|
Q(park__location__coordinates__distance_lte=(point, Distance(km=distance_km)))
|
|
).select_related(
|
|
'park',
|
|
'manufacturer'
|
|
).prefetch_related(
|
|
'park__location'
|
|
).distance(point).order_by('distance')[:limit]
|
|
|
|
return results
|
|
|
|
|
|
def search_all_locations(*, query: str, limit: int = 20) -> Dict[str, QuerySet]:
|
|
"""
|
|
Search across all location types for a query string.
|
|
|
|
Args:
|
|
query: Search string
|
|
limit: Maximum results per type
|
|
|
|
Returns:
|
|
Dictionary containing search results by type
|
|
"""
|
|
results = {}
|
|
|
|
# Search parks
|
|
results['parks'] = Park.objects.filter(
|
|
Q(name__icontains=query) |
|
|
Q(description__icontains=query) |
|
|
Q(location__city__icontains=query) |
|
|
Q(location__region__icontains=query)
|
|
).select_related(
|
|
'operator'
|
|
).prefetch_related(
|
|
'location'
|
|
).order_by('name')[:limit]
|
|
|
|
# Search rides
|
|
results['rides'] = Ride.objects.filter(
|
|
Q(name__icontains=query) |
|
|
Q(description__icontains=query) |
|
|
Q(park__name__icontains=query) |
|
|
Q(manufacturer__name__icontains=query)
|
|
).select_related(
|
|
'park',
|
|
'manufacturer'
|
|
).prefetch_related(
|
|
'park__location'
|
|
).order_by('park__name', 'name')[:limit]
|
|
|
|
return results
|
|
|
|
|
|
def page_views_for_analytics(
|
|
*,
|
|
start_date: Optional[timezone.datetime] = None,
|
|
end_date: Optional[timezone.datetime] = None,
|
|
path_pattern: Optional[str] = None
|
|
) -> QuerySet[PageView]:
|
|
"""
|
|
Get page views for analytics with optional filtering.
|
|
|
|
Args:
|
|
start_date: Start date for filtering
|
|
end_date: End date for filtering
|
|
path_pattern: URL path pattern to filter by
|
|
|
|
Returns:
|
|
QuerySet of page views
|
|
"""
|
|
queryset = PageView.objects.all()
|
|
|
|
if start_date:
|
|
queryset = queryset.filter(timestamp__gte=start_date)
|
|
|
|
if end_date:
|
|
queryset = queryset.filter(timestamp__lte=end_date)
|
|
|
|
if path_pattern:
|
|
queryset = queryset.filter(path__icontains=path_pattern)
|
|
|
|
return queryset.order_by('-timestamp')
|
|
|
|
|
|
def popular_pages_summary(*, days: int = 30) -> Dict[str, Any]:
|
|
"""
|
|
Get summary of most popular pages in the last N days.
|
|
|
|
Args:
|
|
days: Number of days to analyze
|
|
|
|
Returns:
|
|
Dictionary containing popular pages statistics
|
|
"""
|
|
cutoff_date = timezone.now() - timedelta(days=days)
|
|
|
|
# Most viewed pages
|
|
popular_pages = PageView.objects.filter(
|
|
timestamp__gte=cutoff_date
|
|
).values('path').annotate(
|
|
view_count=Count('id')
|
|
).order_by('-view_count')[:10]
|
|
|
|
# Total page views
|
|
total_views = PageView.objects.filter(
|
|
timestamp__gte=cutoff_date
|
|
).count()
|
|
|
|
# Unique visitors (based on IP)
|
|
unique_visitors = PageView.objects.filter(
|
|
timestamp__gte=cutoff_date
|
|
).values('ip_address').distinct().count()
|
|
|
|
return {
|
|
'popular_pages': list(popular_pages),
|
|
'total_views': total_views,
|
|
'unique_visitors': unique_visitors,
|
|
'period_days': days
|
|
}
|
|
|
|
|
|
def geographic_distribution_summary() -> Dict[str, Any]:
|
|
"""
|
|
Get geographic distribution statistics for all locations.
|
|
|
|
Returns:
|
|
Dictionary containing geographic statistics
|
|
"""
|
|
# Parks by country
|
|
parks_by_country = Park.objects.filter(
|
|
location__country__isnull=False
|
|
).values('location__country').annotate(
|
|
count=Count('id')
|
|
).order_by('-count')
|
|
|
|
# Rides by country (through park location)
|
|
rides_by_country = Ride.objects.filter(
|
|
park__location__country__isnull=False
|
|
).values('park__location__country').annotate(
|
|
count=Count('id')
|
|
).order_by('-count')
|
|
|
|
return {
|
|
'parks_by_country': list(parks_by_country),
|
|
'rides_by_country': list(rides_by_country)
|
|
}
|
|
|
|
|
|
def system_health_metrics() -> Dict[str, Any]:
|
|
"""
|
|
Get system health and activity metrics.
|
|
|
|
Returns:
|
|
Dictionary containing system health statistics
|
|
"""
|
|
now = timezone.now()
|
|
last_24h = now - timedelta(hours=24)
|
|
last_7d = now - timedelta(days=7)
|
|
|
|
return {
|
|
'total_parks': Park.objects.count(),
|
|
'operating_parks': Park.objects.filter(status='OPERATING').count(),
|
|
'total_rides': Ride.objects.count(),
|
|
'page_views_24h': PageView.objects.filter(timestamp__gte=last_24h).count(),
|
|
'page_views_7d': PageView.objects.filter(timestamp__gte=last_7d).count(),
|
|
'data_freshness': {
|
|
'latest_park_update': Park.objects.order_by('-updated_at').first().updated_at if Park.objects.exists() else None,
|
|
'latest_ride_update': Ride.objects.order_by('-updated_at').first().updated_at if Ride.objects.exists() else None,
|
|
}
|
|
}
|