""" Park endpoints for API v1. Provides CRUD operations for Park entities with filtering, search, and geographic queries. Supports both SQLite (lat/lng) and PostGIS (location_point) modes. """ from typing import List, Optional from uuid import UUID from decimal import Decimal from django.shortcuts import get_object_or_404 from django.db.models import Q from django.conf import settings from ninja import Router, Query from ninja.pagination import paginate, PageNumberPagination import math from apps.entities.models import Park, Company, _using_postgis from ..schemas import ( ParkCreate, ParkUpdate, ParkOut, ParkListOut, ErrorResponse ) router = Router(tags=["Parks"]) class ParkPagination(PageNumberPagination): """Custom pagination for parks.""" page_size = 50 @router.get( "/", response={200: List[ParkOut]}, summary="List parks", description="Get a paginated list of parks with optional filtering" ) @paginate(ParkPagination) def list_parks( request, search: Optional[str] = Query(None, description="Search by park name"), park_type: Optional[str] = Query(None, description="Filter by park type"), status: Optional[str] = Query(None, description="Filter by status"), operator_id: Optional[UUID] = Query(None, description="Filter by operator"), ordering: Optional[str] = Query("-created", description="Sort by field (prefix with - for descending)") ): """ List all parks with optional filters. **Filters:** - search: Search park names (case-insensitive partial match) - park_type: Filter by park type - status: Filter by operational status - operator_id: Filter by operator company - ordering: Sort results (default: -created) **Returns:** Paginated list of parks """ queryset = Park.objects.select_related('operator').all() # Apply search filter if search: queryset = queryset.filter( Q(name__icontains=search) | Q(description__icontains=search) ) # Apply park type filter if park_type: queryset = queryset.filter(park_type=park_type) # Apply status filter if status: queryset = queryset.filter(status=status) # Apply operator filter if operator_id: queryset = queryset.filter(operator_id=operator_id) # Apply ordering valid_order_fields = ['name', 'created', 'modified', 'opening_date', 'ride_count', 'coaster_count'] order_field = ordering.lstrip('-') if order_field in valid_order_fields: queryset = queryset.order_by(ordering) else: queryset = queryset.order_by('-created') # Annotate with operator name for park in queryset: park.operator_name = park.operator.name if park.operator else None return queryset @router.get( "/{park_id}", response={200: ParkOut, 404: ErrorResponse}, summary="Get park", description="Retrieve a single park by ID" ) def get_park(request, park_id: UUID): """ Get a park by ID. **Parameters:** - park_id: UUID of the park **Returns:** Park details """ park = get_object_or_404(Park.objects.select_related('operator'), id=park_id) park.operator_name = park.operator.name if park.operator else None park.coordinates = park.coordinates return park @router.get( "/nearby/", response={200: List[ParkOut]}, summary="Find nearby parks", description="Find parks within a radius of given coordinates. Uses PostGIS in production, bounding box in SQLite." ) def find_nearby_parks( request, latitude: float = Query(..., description="Latitude coordinate"), longitude: float = Query(..., description="Longitude coordinate"), radius: float = Query(50, description="Search radius in kilometers"), limit: int = Query(50, description="Maximum number of results") ): """ Find parks near a geographic point. **Geographic Search Modes:** - **PostGIS (Production)**: Uses accurate distance-based search with location_point field - **SQLite (Local Dev)**: Uses bounding box approximation with latitude/longitude fields **Parameters:** - latitude: Center point latitude - longitude: Center point longitude - radius: Search radius in kilometers (default: 50) - limit: Maximum results to return (default: 50) **Returns:** List of nearby parks """ if _using_postgis: # Use PostGIS for accurate distance-based search try: from django.contrib.gis.measure import D from django.contrib.gis.geos import Point user_point = Point(longitude, latitude, srid=4326) nearby_parks = Park.objects.filter( location_point__distance_lte=(user_point, D(km=radius)) ).select_related('operator')[:limit] except Exception as e: return {"detail": f"Geographic search error: {str(e)}"}, 500 else: # Use bounding box approximation for SQLite # Calculate rough bounding box (1 degree ≈ 111 km at equator) lat_offset = radius / 111.0 lng_offset = radius / (111.0 * math.cos(math.radians(latitude))) min_lat = latitude - lat_offset max_lat = latitude + lat_offset min_lng = longitude - lng_offset max_lng = longitude + lng_offset nearby_parks = Park.objects.filter( latitude__gte=Decimal(str(min_lat)), latitude__lte=Decimal(str(max_lat)), longitude__gte=Decimal(str(min_lng)), longitude__lte=Decimal(str(max_lng)) ).select_related('operator')[:limit] # Annotate results results = [] for park in nearby_parks: park.operator_name = park.operator.name if park.operator else None park.coordinates = park.coordinates results.append(park) return results @router.post( "/", response={201: ParkOut, 400: ErrorResponse}, summary="Create park", description="Create a new park (requires authentication)" ) def create_park(request, payload: ParkCreate): """ Create a new park. **Authentication:** Required **Parameters:** - payload: Park data **Returns:** Created park """ # TODO: Add authentication check # if not request.auth: # return 401, {"detail": "Authentication required"} data = payload.dict() # Extract coordinates to use set_location method latitude = data.pop('latitude', None) longitude = data.pop('longitude', None) park = Park.objects.create(**data) # Set location using helper method (handles both SQLite and PostGIS) if latitude is not None and longitude is not None: park.set_location(longitude, latitude) park.save() park.coordinates = park.coordinates if park.operator: park.operator_name = park.operator.name return 201, park @router.put( "/{park_id}", response={200: ParkOut, 404: ErrorResponse, 400: ErrorResponse}, summary="Update park", description="Update an existing park (requires authentication)" ) def update_park(request, park_id: UUID, payload: ParkUpdate): """ Update a park. **Authentication:** Required **Parameters:** - park_id: UUID of the park - payload: Updated park data **Returns:** Updated park """ # TODO: Add authentication check # if not request.auth: # return 401, {"detail": "Authentication required"} park = get_object_or_404(Park.objects.select_related('operator'), id=park_id) data = payload.dict(exclude_unset=True) # Handle coordinates separately latitude = data.pop('latitude', None) longitude = data.pop('longitude', None) # Update other fields for key, value in data.items(): setattr(park, key, value) # Update location if coordinates provided if latitude is not None and longitude is not None: park.set_location(longitude, latitude) park.save() park.operator_name = park.operator.name if park.operator else None park.coordinates = park.coordinates return park @router.patch( "/{park_id}", response={200: ParkOut, 404: ErrorResponse, 400: ErrorResponse}, summary="Partial update park", description="Partially update an existing park (requires authentication)" ) def partial_update_park(request, park_id: UUID, payload: ParkUpdate): """ Partially update a park. **Authentication:** Required **Parameters:** - park_id: UUID of the park - payload: Fields to update **Returns:** Updated park """ # TODO: Add authentication check # if not request.auth: # return 401, {"detail": "Authentication required"} park = get_object_or_404(Park.objects.select_related('operator'), id=park_id) data = payload.dict(exclude_unset=True) # Handle coordinates separately latitude = data.pop('latitude', None) longitude = data.pop('longitude', None) # Update other fields for key, value in data.items(): setattr(park, key, value) # Update location if coordinates provided if latitude is not None and longitude is not None: park.set_location(longitude, latitude) park.save() park.operator_name = park.operator.name if park.operator else None park.coordinates = park.coordinates return park @router.delete( "/{park_id}", response={204: None, 404: ErrorResponse}, summary="Delete park", description="Delete a park (requires authentication)" ) def delete_park(request, park_id: UUID): """ Delete a park. **Authentication:** Required **Parameters:** - park_id: UUID of the park **Returns:** No content (204) """ # TODO: Add authentication check # if not request.auth: # return 401, {"detail": "Authentication required"} park = get_object_or_404(Park, id=park_id) park.delete() return 204, None @router.get( "/{park_id}/rides", response={200: List[dict], 404: ErrorResponse}, summary="Get park rides", description="Get all rides at a park" ) def get_park_rides(request, park_id: UUID): """ Get all rides at a park. **Parameters:** - park_id: UUID of the park **Returns:** List of rides """ park = get_object_or_404(Park, id=park_id) rides = park.rides.select_related('manufacturer').all().values( 'id', 'name', 'slug', 'status', 'ride_category', 'is_coaster', 'manufacturer__name' ) return list(rides)