pacnpal/thrilltrack-explorer
1
0
Fork 0
mirror of https://github.com/pacnpal/thrilltrack-explorer.git synced 2025-12-23 06:51:13 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add ride credits and top lists endpoints for API v1

Browse Source 
- Implement CRUD operations for ride credits, allowing users to log rides, track counts, and view statistics.
- Create endpoints for managing user-created ranked lists of parks, rides, or coasters with custom rankings and notes.
- Introduce pagination for both ride credits and top lists.
- Ensure proper authentication and authorization for modifying user-specific data.
- Add serialization methods for ride credits and top lists to return structured data.
- Include error handling and logging for better traceability of operations.
This commit is contained in:
pacnpal
2025-11-08 16:02:11 -05:00
parent 00985eac8d
commit e38a9aaa41
11 changed files with 2176 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
django/api/v1/__pycache__/api.cpython-313.pyc
View File

Binary file not shown.

BIN
django/api/v1/__pycache__/schemas.cpython-313.pyc
View File

Binary file not shown.

8
django/api/v1/api.py
View File

@@ -15,6 +15,9 @@ from .endpoints.versioning import router as versioning_router
from .endpoints.auth import router as auth_router
from .endpoints.photos import router as photos_router
from .endpoints.search import router as search_router
from .endpoints.reviews import router as reviews_router
from .endpoints.ride_credits import router as ride_credits_router
from .endpoints.top_lists import router as top_lists_router
# Create the main API instance
@@ -107,6 +110,11 @@ api.add_router("", photos_router) # Photos endpoints include both /photos and e
# Add search router
api.add_router("/search", search_router)
# Add user interaction routers
api.add_router("/reviews", reviews_router)
api.add_router("/ride-credits", ride_credits_router)
api.add_router("/top-lists", top_lists_router)
# Health check endpoint
@api.get("/health", tags=["System"], summary="Health check")

BIN
django/api/v1/endpoints/__pycache__/reviews.cpython-313.pyc Normal file
View File

Binary file not shown.

BIN
django/api/v1/endpoints/__pycache__/ride_credits.cpython-313.pyc Normal file
View File

Binary file not shown.

BIN
django/api/v1/endpoints/__pycache__/top_lists.cpython-313.pyc Normal file
View File

Binary file not shown.

586
django/api/v1/endpoints/reviews.py Normal file
View File

@@ -0,0 +1,586 @@
"""
Review endpoints for API v1.
Provides CRUD operations for reviews with moderation workflow integration.
Users can review parks and rides, vote on reviews, and moderators can approve/reject.
"""
from typing import List, Optional
from uuid import UUID
from django.shortcuts import get_object_or_404
from django.db.models import Q, Count, Avg
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from ninja import Router, Query
from ninja.pagination import paginate, PageNumberPagination
import logging
from apps.reviews.models import Review, ReviewHelpfulVote
from apps.entities.models import Park, Ride
from apps.users.permissions import jwt_auth, require_auth
from ..schemas import (
ReviewCreateSchema,
ReviewUpdateSchema,
ReviewOut,
ReviewListOut,
ReviewStatsOut,
VoteRequest,
VoteResponse,
ErrorResponse,
UserSchema,
)
router = Router(tags=["Reviews"])
logger = logging.getLogger(__name__)
class ReviewPagination(PageNumberPagination):
"""Custom pagination for reviews."""
page_size = 50
def _get_entity(entity_type: str, entity_id: UUID):
"""Helper to get and validate entity (Park or Ride)."""
if entity_type == 'park':
return get_object_or_404(Park, id=entity_id), ContentType.objects.get_for_model(Park)
elif entity_type == 'ride':
return get_object_or_404(Ride, id=entity_id), ContentType.objects.get_for_model(Ride)
else:
raise ValidationError(f"Invalid entity_type: {entity_type}")
def _serialize_review(review: Review, user=None) -> dict:
"""Serialize review with computed fields."""
data = {
'id': review.id,
'user': UserSchema(
id=review.user.id,
username=review.user.username,
display_name=review.user.display_name,
avatar_url=review.user.avatar_url,
reputation_score=review.user.reputation_score,
),
'entity_type': review.content_type.model,
'entity_id': str(review.object_id),
'entity_name': str(review.content_object) if review.content_object else 'Unknown',
'title': review.title,
'content': review.content,
'rating': review.rating,
'visit_date': review.visit_date,
'wait_time_minutes': review.wait_time_minutes,
'helpful_votes': review.helpful_votes,
'total_votes': review.total_votes,
'helpful_percentage': review.helpful_percentage,
'moderation_status': review.moderation_status,
'moderated_at': review.moderated_at,
'moderated_by_email': review.moderated_by.email if review.moderated_by else None,
'photo_count': review.photos.count(),
'created': review.created,
'modified': review.modified,
'user_vote': None,
}
# Add user's vote if authenticated
if user and user.is_authenticated:
try:
vote = ReviewHelpfulVote.objects.get(review=review, user=user)
data['user_vote'] = vote.is_helpful
except ReviewHelpfulVote.DoesNotExist:
pass
return data
# ============================================================================
# Main Review CRUD Endpoints
# ============================================================================
@router.post("/", response={201: ReviewOut, 400: ErrorResponse, 409: ErrorResponse}, auth=jwt_auth)
@require_auth
def create_review(request, data: ReviewCreateSchema):
"""
Create a new review for a park or ride.
**Authentication:** Required
**Parameters:**
- entity_type: "park" or "ride"
- entity_id: UUID of park or ride
- title: Review title
- content: Review content (min 10 characters)
- rating: 1-5 stars
- visit_date: Optional visit date
- wait_time_minutes: Optional wait time
**Returns:** Created review (pending moderation)
**Note:** Reviews automatically enter moderation workflow.
Users can only create one review per entity.
"""
try:
user = request.auth
# Get and validate entity
entity, content_type = _get_entity(data.entity_type, data.entity_id)
# Check for duplicate review
existing = Review.objects.filter(
user=user,
content_type=content_type,
object_id=entity.id
).first()
if existing:
return 409, {
'detail': f"You have already reviewed this {data.entity_type}. "
f"Use PUT /reviews/{existing.id}/ to update your review."
}
# Create review
review = Review.objects.create(
user=user,
content_type=content_type,
object_id=entity.id,
title=data.title,
content=data.content,
rating=data.rating,
visit_date=data.visit_date,
wait_time_minutes=data.wait_time_minutes,
moderation_status=Review.MODERATION_PENDING,
)
logger.info(f"Review created: {review.id} by {user.email} for {data.entity_type} {entity.id}")
# Serialize and return
review_data = _serialize_review(review, user)
return 201, review_data
except ValidationError as e:
return 400, {'detail': str(e)}
except Exception as e:
logger.error(f"Error creating review: {e}")
return 400, {'detail': str(e)}
@router.get("/", response={200: List[ReviewOut]})
@paginate(ReviewPagination)
def list_reviews(
request,
entity_type: Optional[str] = Query(None, description="Filter by entity type: park or ride"),
entity_id: Optional[UUID] = Query(None, description="Filter by specific entity ID"),
user_id: Optional[UUID] = Query(None, description="Filter by user ID"),
rating: Optional[int] = Query(None, ge=1, le=5, description="Filter by rating"),
moderation_status: Optional[str] = Query(None, description="Filter by moderation status"),
ordering: Optional[str] = Query("-created", description="Sort by field")
):
"""
List reviews with optional filtering.
**Authentication:** Optional (only approved reviews shown if not authenticated/not moderator)
**Filters:**
- entity_type: park or ride
- entity_id: Specific park/ride
- user_id: Reviews by specific user
- rating: Filter by star rating
- moderation_status: pending/approved/rejected (moderators only)
- ordering: Sort field (default: -created)
**Returns:** Paginated list of reviews
"""
# Base query with optimizations
queryset = Review.objects.select_related(
'user',
'moderated_by',
'content_type'
).prefetch_related('photos')
# Check if user is authenticated and is moderator
user = request.auth if hasattr(request, 'auth') else None
is_moderator = user and hasattr(user, 'role') and user.role.is_moderator if user else False
# Apply moderation filter
if not is_moderator:
queryset = queryset.filter(moderation_status=Review.MODERATION_APPROVED)
# Apply entity type filter
if entity_type:
if entity_type == 'park':
ct = ContentType.objects.get_for_model(Park)
elif entity_type == 'ride':
ct = ContentType.objects.get_for_model(Ride)
else:
queryset = queryset.none()
queryset = queryset.filter(content_type=ct)
# Apply entity ID filter
if entity_id:
queryset = queryset.filter(object_id=entity_id)
# Apply user filter
if user_id:
queryset = queryset.filter(user_id=user_id)
# Apply rating filter
if rating:
queryset = queryset.filter(rating=rating)
# Apply moderation status filter (moderators only)
if moderation_status and is_moderator:
queryset = queryset.filter(moderation_status=moderation_status)
# Apply ordering
valid_order_fields = ['created', 'modified', 'rating', 'helpful_votes', 'visit_date']
order_field = ordering.lstrip('-')
if order_field in valid_order_fields:
queryset = queryset.order_by(ordering)
else:
queryset = queryset.order_by('-created')
# Serialize reviews
reviews = [_serialize_review(review, user) for review in queryset]
return reviews
@router.get("/{review_id}", response={200: ReviewOut, 404: ErrorResponse})
def get_review(request, review_id: int):
"""
Get a specific review by ID.
**Authentication:** Optional
**Parameters:**
- review_id: Review ID
**Returns:** Review details
**Note:** Only approved reviews are accessible to non-moderators.
"""
user = request.auth if hasattr(request, 'auth') else None
is_moderator = user and hasattr(user, 'role') and user.role.is_moderator if user else False
is_owner = user and Review.objects.filter(id=review_id, user=user).exists() if user else False
review = get_object_or_404(
Review.objects.select_related('user', 'moderated_by', 'content_type').prefetch_related('photos'),
id=review_id
)
# Check access
if not review.is_approved and not is_moderator and not is_owner:
return 404, {'detail': 'Review not found'}
review_data = _serialize_review(review, user)
return 200, review_data
@router.put("/{review_id}", response={200: ReviewOut, 403: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def update_review(request, review_id: int, data: ReviewUpdateSchema):
"""
Update your own review.
**Authentication:** Required (must be review owner)
**Parameters:**
- review_id: Review ID
- data: Fields to update
**Returns:** Updated review
**Note:** Updating a review resets it to pending moderation.
"""
user = request.auth
review = get_object_or_404(
Review.objects.select_related('user', 'content_type'),
id=review_id
)
# Check ownership
if review.user != user:
return 403, {'detail': 'You can only update your own reviews'}
# Update fields
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(review, key, value)
# Reset to pending moderation
review.moderation_status = Review.MODERATION_PENDING
review.moderated_at = None
review.moderated_by = None
review.moderation_notes = ''
review.save()
logger.info(f"Review updated: {review.id} by {user.email}")
review_data = _serialize_review(review, user)
return 200, review_data
@router.delete("/{review_id}", response={204: None, 403: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def delete_review(request, review_id: int):
"""
Delete your own review.
**Authentication:** Required (must be review owner)
**Parameters:**
- review_id: Review ID
**Returns:** No content (204)
"""
user = request.auth
review = get_object_or_404(Review, id=review_id)
# Check ownership
if review.user != user:
return 403, {'detail': 'You can only delete your own reviews'}
logger.info(f"Review deleted: {review.id} by {user.email}")
review.delete()
return 204, None
# ============================================================================
# Voting Endpoint
# ============================================================================
@router.post("/{review_id}/vote", response={200: VoteResponse, 400: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def vote_on_review(request, review_id: int, data: VoteRequest):
"""
Vote on a review (helpful or not helpful).
**Authentication:** Required
**Parameters:**
- review_id: Review ID
- is_helpful: True if helpful, False if not helpful
**Returns:** Updated vote counts
**Note:** Users can change their vote but cannot vote on their own reviews.
"""
user = request.auth
review = get_object_or_404(Review, id=review_id)
# Prevent self-voting
if review.user == user:
return 400, {'detail': 'You cannot vote on your own review'}
# Create or update vote
vote, created = ReviewHelpfulVote.objects.update_or_create(
review=review,
user=user,
defaults={'is_helpful': data.is_helpful}
)
# Refresh review to get updated counts
review.refresh_from_db()
return 200, {
'success': True,
'review_id': review.id,
'helpful_votes': review.helpful_votes,
'total_votes': review.total_votes,
'helpful_percentage': review.helpful_percentage,
}
# ============================================================================
# Entity-Specific Review Endpoints
# ============================================================================
@router.get("/parks/{park_id}", response={200: List[ReviewOut]})
@paginate(ReviewPagination)
def get_park_reviews(
request,
park_id: UUID,
rating: Optional[int] = Query(None, ge=1, le=5),
ordering: Optional[str] = Query("-created")
):
"""
Get all reviews for a specific park.
**Parameters:**
- park_id: Park UUID
- rating: Optional rating filter
- ordering: Sort field (default: -created)
**Returns:** Paginated list of park reviews
"""
park = get_object_or_404(Park, id=park_id)
content_type = ContentType.objects.get_for_model(Park)
user = request.auth if hasattr(request, 'auth') else None
is_moderator = user and hasattr(user, 'role') and user.role.is_moderator if user else False
queryset = Review.objects.filter(
content_type=content_type,
object_id=park.id
).select_related('user', 'moderated_by').prefetch_related('photos')
if not is_moderator:
queryset = queryset.filter(moderation_status=Review.MODERATION_APPROVED)
if rating:
queryset = queryset.filter(rating=rating)
valid_order_fields = ['created', 'modified', 'rating', 'helpful_votes', 'visit_date']
order_field = ordering.lstrip('-')
if order_field in valid_order_fields:
queryset = queryset.order_by(ordering)
else:
queryset = queryset.order_by('-created')
reviews = [_serialize_review(review, user) for review in queryset]
return reviews
@router.get("/rides/{ride_id}", response={200: List[ReviewOut]})
@paginate(ReviewPagination)
def get_ride_reviews(
request,
ride_id: UUID,
rating: Optional[int] = Query(None, ge=1, le=5),
ordering: Optional[str] = Query("-created")
):
"""
Get all reviews for a specific ride.
**Parameters:**
- ride_id: Ride UUID
- rating: Optional rating filter
- ordering: Sort field (default: -created)
**Returns:** Paginated list of ride reviews
"""
ride = get_object_or_404(Ride, id=ride_id)
content_type = ContentType.objects.get_for_model(Ride)
user = request.auth if hasattr(request, 'auth') else None
is_moderator = user and hasattr(user, 'role') and user.role.is_moderator if user else False
queryset = Review.objects.filter(
content_type=content_type,
object_id=ride.id
).select_related('user', 'moderated_by').prefetch_related('photos')
if not is_moderator:
queryset = queryset.filter(moderation_status=Review.MODERATION_APPROVED)
if rating:
queryset = queryset.filter(rating=rating)
valid_order_fields = ['created', 'modified', 'rating', 'helpful_votes', 'visit_date']
order_field = ordering.lstrip('-')
if order_field in valid_order_fields:
queryset = queryset.order_by(ordering)
else:
queryset = queryset.order_by('-created')
reviews = [_serialize_review(review, user) for review in queryset]
return reviews
@router.get("/users/{user_id}", response={200: List[ReviewOut]})
@paginate(ReviewPagination)
def get_user_reviews(
request,
user_id: UUID,
entity_type: Optional[str] = Query(None),
ordering: Optional[str] = Query("-created")
):
"""
Get all reviews by a specific user.
**Parameters:**
- user_id: User UUID
- entity_type: Optional filter (park or ride)
- ordering: Sort field (default: -created)
**Returns:** Paginated list of user's reviews
**Note:** Only approved reviews visible unless viewing own reviews or moderator.
"""
user = request.auth if hasattr(request, 'auth') else None
is_owner = user and str(user.id) == str(user_id) if user else False
is_moderator = user and hasattr(user, 'role') and user.role.is_moderator if user else False
queryset = Review.objects.filter(
user_id=user_id
).select_related('user', 'moderated_by', 'content_type').prefetch_related('photos')
# Filter by moderation status
if not is_owner and not is_moderator:
queryset = queryset.filter(moderation_status=Review.MODERATION_APPROVED)
# Apply entity type filter
if entity_type:
if entity_type == 'park':
ct = ContentType.objects.get_for_model(Park)
elif entity_type == 'ride':
ct = ContentType.objects.get_for_model(Ride)
else:
queryset = queryset.none()
queryset = queryset.filter(content_type=ct)
# Apply ordering
valid_order_fields = ['created', 'modified', 'rating', 'helpful_votes', 'visit_date']
order_field = ordering.lstrip('-')
if order_field in valid_order_fields:
queryset = queryset.order_by(ordering)
else:
queryset = queryset.order_by('-created')
reviews = [_serialize_review(review, user) for review in queryset]
return reviews
# ============================================================================
# Statistics Endpoint
# ============================================================================
@router.get("/stats/{entity_type}/{entity_id}", response={200: ReviewStatsOut, 404: ErrorResponse})
def get_review_stats(request, entity_type: str, entity_id: UUID):
"""
Get review statistics for a park or ride.
**Parameters:**
- entity_type: "park" or "ride"
- entity_id: Entity UUID
**Returns:** Statistics including average rating and distribution
"""
try:
entity, content_type = _get_entity(entity_type, entity_id)
except ValidationError as e:
return 404, {'detail': str(e)}
# Get approved reviews only
reviews = Review.objects.filter(
content_type=content_type,
object_id=entity.id,
moderation_status=Review.MODERATION_APPROVED
)
# Calculate stats
stats = reviews.aggregate(
average_rating=Avg('rating'),
total_reviews=Count('id')
)
# Get rating distribution
distribution = {}
for rating in range(1, 6):
distribution[rating] = reviews.filter(rating=rating).count()
return 200, {
'average_rating': stats['average_rating'] or 0.0,
'total_reviews': stats['total_reviews'] or 0,
'rating_distribution': distribution,
}

410
django/api/v1/endpoints/ride_credits.py Normal file
View File

@@ -0,0 +1,410 @@
"""
Ride Credit endpoints for API v1.
Provides CRUD operations for tracking which rides users have ridden (coaster counting).
Users can log rides, track ride counts, and view statistics.
"""
from typing import List, Optional
from uuid import UUID
from datetime import date
from django.shortcuts import get_object_or_404
from django.db.models import Count, Sum, Min, Max, Q
from ninja import Router, Query
from ninja.pagination import paginate, PageNumberPagination
import logging
from apps.users.models import UserRideCredit, User
from apps.entities.models import Ride
from apps.users.permissions import jwt_auth, require_auth
from ..schemas import (
RideCreditCreateSchema,
RideCreditUpdateSchema,
RideCreditOut,
RideCreditListOut,
RideCreditStatsOut,
ErrorResponse,
UserSchema,
)
router = Router(tags=["Ride Credits"])
logger = logging.getLogger(__name__)
class RideCreditPagination(PageNumberPagination):
"""Custom pagination for ride credits."""
page_size = 50
def _serialize_ride_credit(credit: UserRideCredit) -> dict:
"""Serialize ride credit with computed fields."""
ride = credit.ride
park = ride.park
return {
'id': credit.id,
'user': UserSchema(
id=credit.user.id,
username=credit.user.username,
display_name=credit.user.display_name,
avatar_url=credit.user.avatar_url,
reputation_score=credit.user.reputation_score,
),
'ride_id': str(ride.id),
'ride_name': ride.name,
'ride_slug': ride.slug,
'park_id': str(park.id),
'park_name': park.name,
'park_slug': park.slug,
'is_coaster': ride.is_coaster,
'first_ride_date': credit.first_ride_date,
'ride_count': credit.ride_count,
'notes': credit.notes or '',
'created': credit.created,
'modified': credit.modified,
}
# ============================================================================
# Main Ride Credit CRUD Endpoints
# ============================================================================
@router.post("/", response={201: RideCreditOut, 400: ErrorResponse}, auth=jwt_auth)
@require_auth
def create_ride_credit(request, data: RideCreditCreateSchema):
"""
Log a ride (create or update ride credit).
**Authentication:** Required
**Parameters:**
- ride_id: UUID of ride
- first_ride_date: Date of first ride (optional)
- ride_count: Number of times ridden (default: 1)
- notes: Notes about the ride experience (optional)
**Returns:** Created or updated ride credit
**Note:** If a credit already exists, it updates the ride_count.
"""
try:
user = request.auth
# Validate ride exists
ride = get_object_or_404(Ride, id=data.ride_id)
# Check if credit already exists
credit, created = UserRideCredit.objects.get_or_create(
user=user,
ride=ride,
defaults={
'first_ride_date': data.first_ride_date,
'ride_count': data.ride_count,
'notes': data.notes or '',
}
)
if not created:
# Update existing credit
credit.ride_count += data.ride_count
if data.first_ride_date and (not credit.first_ride_date or data.first_ride_date < credit.first_ride_date):
credit.first_ride_date = data.first_ride_date
if data.notes:
credit.notes = data.notes
credit.save()
logger.info(f"Ride credit {'created' if created else 'updated'}: {credit.id} by {user.email}")
credit_data = _serialize_ride_credit(credit)
return 201, credit_data
except Exception as e:
logger.error(f"Error creating ride credit: {e}")
return 400, {'detail': str(e)}
@router.get("/", response={200: List[RideCreditOut]}, auth=jwt_auth)
@require_auth
@paginate(RideCreditPagination)
def list_my_ride_credits(
request,
ride_id: Optional[UUID] = Query(None, description="Filter by ride"),
park_id: Optional[UUID] = Query(None, description="Filter by park"),
is_coaster: Optional[bool] = Query(None, description="Filter coasters only"),
date_from: Optional[date] = Query(None, description="Credits from date"),
date_to: Optional[date] = Query(None, description="Credits to date"),
ordering: Optional[str] = Query("-first_ride_date", description="Sort by field")
):
"""
List your own ride credits.
**Authentication:** Required
**Filters:**
- ride_id: Specific ride
- park_id: Rides at specific park
- is_coaster: Coasters only
- date_from: Credits from date
- date_to: Credits to date
- ordering: Sort field (default: -first_ride_date)
**Returns:** Paginated list of your ride credits
"""
user = request.auth
# Base query with optimizations
queryset = UserRideCredit.objects.filter(user=user).select_related('ride__park')
# Apply ride filter
if ride_id:
queryset = queryset.filter(ride_id=ride_id)
# Apply park filter
if park_id:
queryset = queryset.filter(ride__park_id=park_id)
# Apply coaster filter
if is_coaster is not None:
queryset = queryset.filter(ride__is_coaster=is_coaster)
# Apply date filters
if date_from:
queryset = queryset.filter(first_ride_date__gte=date_from)
if date_to:
queryset = queryset.filter(first_ride_date__lte=date_to)
# Apply ordering
valid_order_fields = ['first_ride_date', 'ride_count', 'created', 'modified']
order_field = ordering.lstrip('-')
if order_field in valid_order_fields:
queryset = queryset.order_by(ordering)
else:
queryset = queryset.order_by('-first_ride_date')
# Serialize credits
credits = [_serialize_ride_credit(credit) for credit in queryset]
return credits
@router.get("/{credit_id}", response={200: RideCreditOut, 403: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def get_ride_credit(request, credit_id: UUID):
"""
Get a specific ride credit by ID.
**Authentication:** Required (must be credit owner)
**Parameters:**
- credit_id: Credit UUID
**Returns:** Credit details
"""
user = request.auth
credit = get_object_or_404(
UserRideCredit.objects.select_related('ride__park'),
id=credit_id
)
# Check ownership
if credit.user != user:
return 403, {'detail': 'You can only view your own ride credits'}
credit_data = _serialize_ride_credit(credit)
return 200, credit_data
@router.put("/{credit_id}", response={200: RideCreditOut, 403: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def update_ride_credit(request, credit_id: UUID, data: RideCreditUpdateSchema):
"""
Update a ride credit.
**Authentication:** Required (must be credit owner)
**Parameters:**
- credit_id: Credit UUID
- data: Fields to update
**Returns:** Updated credit
"""
user = request.auth
credit = get_object_or_404(
UserRideCredit.objects.select_related('ride__park'),
id=credit_id
)
# Check ownership
if credit.user != user:
return 403, {'detail': 'You can only update your own ride credits'}
# Update fields
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(credit, key, value)
credit.save()
logger.info(f"Ride credit updated: {credit.id} by {user.email}")
credit_data = _serialize_ride_credit(credit)
return 200, credit_data
@router.delete("/{credit_id}", response={204: None, 403: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def delete_ride_credit(request, credit_id: UUID):
"""
Delete a ride credit.
**Authentication:** Required (must be credit owner)
**Parameters:**
- credit_id: Credit UUID
**Returns:** No content (204)
"""
user = request.auth
credit = get_object_or_404(UserRideCredit, id=credit_id)
# Check ownership
if credit.user != user:
return 403, {'detail': 'You can only delete your own ride credits'}
logger.info(f"Ride credit deleted: {credit.id} by {user.email}")
credit.delete()
return 204, None
# ============================================================================
# User-Specific Endpoints
# ============================================================================
@router.get("/users/{user_id}", response={200: List[RideCreditOut], 403: ErrorResponse})
@paginate(RideCreditPagination)
def get_user_ride_credits(
request,
user_id: UUID,
park_id: Optional[UUID] = Query(None),
is_coaster: Optional[bool] = Query(None),
ordering: Optional[str] = Query("-first_ride_date")
):
"""
Get a user's ride credits.
**Authentication:** Optional (respects privacy settings)
**Parameters:**
- user_id: User UUID
- park_id: Filter by park (optional)
- is_coaster: Filter coasters only (optional)
- ordering: Sort field (default: -first_ride_date)
**Returns:** Paginated list of user's ride credits
**Note:** Only visible if user's profile is public or viewer is the owner.
"""
target_user = get_object_or_404(User, id=user_id)
# Check if current user
current_user = request.auth if hasattr(request, 'auth') else None
is_owner = current_user and current_user.id == target_user.id
# Check privacy
if not is_owner:
# Check if profile is public
try:
profile = target_user.profile
if not profile.profile_public:
return 403, {'detail': 'This user\'s ride credits are private'}
except:
return 403, {'detail': 'This user\'s ride credits are private'}
# Build query
queryset = UserRideCredit.objects.filter(user=target_user).select_related('ride__park')
# Apply filters
if park_id:
queryset = queryset.filter(ride__park_id=park_id)
if is_coaster is not None:
queryset = queryset.filter(ride__is_coaster=is_coaster)
# Apply ordering
valid_order_fields = ['first_ride_date', 'ride_count', 'created']
order_field = ordering.lstrip('-')
if order_field in valid_order_fields:
queryset = queryset.order_by(ordering)
else:
queryset = queryset.order_by('-first_ride_date')
# Serialize credits
credits = [_serialize_ride_credit(credit) for credit in queryset]
return credits
@router.get("/users/{user_id}/stats", response={200: RideCreditStatsOut, 403: ErrorResponse})
def get_user_ride_stats(request, user_id: UUID):
"""
Get statistics about a user's ride credits.
**Authentication:** Optional (respects privacy settings)
**Parameters:**
- user_id: User UUID
**Returns:** Statistics including total rides, credits, parks, etc.
"""
target_user = get_object_or_404(User, id=user_id)
# Check if current user
current_user = request.auth if hasattr(request, 'auth') else None
is_owner = current_user and current_user.id == target_user.id
# Check privacy
if not is_owner:
try:
profile = target_user.profile
if not profile.profile_public:
return 403, {'detail': 'This user\'s statistics are private'}
except:
return 403, {'detail': 'This user\'s statistics are private'}
# Get all credits
credits = UserRideCredit.objects.filter(user=target_user).select_related('ride__park')
# Calculate basic stats
stats = credits.aggregate(
total_rides=Sum('ride_count'),
total_credits=Count('id'),
unique_parks=Count('ride__park', distinct=True),
coaster_count=Count('id', filter=Q(ride__is_coaster=True)),
first_credit_date=Min('first_ride_date'),
last_credit_date=Max('first_ride_date'),
)
# Get top park
park_counts = credits.values('ride__park__name').annotate(
count=Count('id')
).order_by('-count').first()
top_park = park_counts['ride__park__name'] if park_counts else None
top_park_count = park_counts['count'] if park_counts else 0
# Get recent credits (last 5)
recent_credits = credits.order_by('-first_ride_date')[:5]
recent_credits_data = [_serialize_ride_credit(c) for c in recent_credits]
return 200, {
'total_rides': stats['total_rides'] or 0,
'total_credits': stats['total_credits'] or 0,
'unique_parks': stats['unique_parks'] or 0,
'coaster_count': stats['coaster_count'] or 0,
'first_credit_date': stats['first_credit_date'],
'last_credit_date': stats['last_credit_date'],
'top_park': top_park,
'top_park_count': top_park_count,
'recent_credits': recent_credits_data,
}

574
django/api/v1/endpoints/top_lists.py Normal file
View File

@@ -0,0 +1,574 @@
"""
Top List endpoints for API v1.
Provides CRUD operations for user-created ranked lists.
Users can create lists of parks, rides, or coasters with custom rankings and notes.
"""
from typing import List, Optional
from uuid import UUID
from django.shortcuts import get_object_or_404
from django.db.models import Q, Max
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.db import transaction
from ninja import Router, Query
from ninja.pagination import paginate, PageNumberPagination
import logging
from apps.users.models import UserTopList, UserTopListItem, User
from apps.entities.models import Park, Ride
from apps.users.permissions import jwt_auth, require_auth
from ..schemas import (
TopListCreateSchema,
TopListUpdateSchema,
TopListItemCreateSchema,
TopListItemUpdateSchema,
TopListOut,
TopListDetailOut,
TopListListOut,
TopListItemOut,
ErrorResponse,
UserSchema,
)
router = Router(tags=["Top Lists"])
logger = logging.getLogger(__name__)
class TopListPagination(PageNumberPagination):
"""Custom pagination for top lists."""
page_size = 50
def _get_entity(entity_type: str, entity_id: UUID):
"""Helper to get and validate entity (Park or Ride)."""
if entity_type == 'park':
return get_object_or_404(Park, id=entity_id), ContentType.objects.get_for_model(Park)
elif entity_type == 'ride':
return get_object_or_404(Ride, id=entity_id), ContentType.objects.get_for_model(Ride)
else:
raise ValidationError(f"Invalid entity_type: {entity_type}")
def _serialize_list_item(item: UserTopListItem) -> dict:
"""Serialize top list item with computed fields."""
entity = item.content_object
data = {
'id': item.id,
'position': item.position,
'entity_type': item.content_type.model,
'entity_id': str(item.object_id),
'entity_name': entity.name if entity else 'Unknown',
'entity_slug': entity.slug if entity and hasattr(entity, 'slug') else '',
'entity_image_url': None, # TODO: Get from entity
'park_name': None,
'notes': item.notes or '',
'created': item.created,
'modified': item.modified,
}
# If entity is a ride, add park name
if item.content_type.model == 'ride' and entity and hasattr(entity, 'park'):
data['park_name'] = entity.park.name if entity.park else None
return data
def _serialize_top_list(top_list: UserTopList, include_items: bool = False) -> dict:
"""Serialize top list with optional items."""
data = {
'id': top_list.id,
'user': UserSchema(
id=top_list.user.id,
username=top_list.user.username,
display_name=top_list.user.display_name,
avatar_url=top_list.user.avatar_url,
reputation_score=top_list.user.reputation_score,
),
'list_type': top_list.list_type,
'title': top_list.title,
'description': top_list.description or '',
'is_public': top_list.is_public,
'item_count': top_list.item_count,
'created': top_list.created,
'modified': top_list.modified,
}
if include_items:
items = top_list.items.select_related('content_type').order_by('position')
data['items'] = [_serialize_list_item(item) for item in items]
return data
# ============================================================================
# Main Top List CRUD Endpoints
# ============================================================================
@router.post("/", response={201: TopListOut, 400: ErrorResponse}, auth=jwt_auth)
@require_auth
def create_top_list(request, data: TopListCreateSchema):
"""
Create a new top list.
**Authentication:** Required
**Parameters:**
- list_type: "parks", "rides", or "coasters"
- title: List title
- description: List description (optional)
- is_public: Whether list is publicly visible (default: true)
**Returns:** Created top list
"""
try:
user = request.auth
# Create list
top_list = UserTopList.objects.create(
user=user,
list_type=data.list_type,
title=data.title,
description=data.description or '',
is_public=data.is_public,
)
logger.info(f"Top list created: {top_list.id} by {user.email}")
list_data = _serialize_top_list(top_list)
return 201, list_data
except Exception as e:
logger.error(f"Error creating top list: {e}")
return 400, {'detail': str(e)}
@router.get("/", response={200: List[TopListOut]})
@paginate(TopListPagination)
def list_top_lists(
request,
list_type: Optional[str] = Query(None, description="Filter by list type"),
user_id: Optional[UUID] = Query(None, description="Filter by user ID"),
ordering: Optional[str] = Query("-created", description="Sort by field")
):
"""
List accessible top lists.
**Authentication:** Optional
**Filters:**
- list_type: parks, rides, or coasters
- user_id: Lists by specific user
- ordering: Sort field (default: -created)
**Returns:** Paginated list of top lists
**Note:** Shows public lists + user's own private lists if authenticated.
"""
user = request.auth if hasattr(request, 'auth') else None
# Base query
queryset = UserTopList.objects.select_related('user')
# Apply visibility filter
if user:
# Show public lists + user's own lists
queryset = queryset.filter(Q(is_public=True) | Q(user=user))
else:
# Only public lists
queryset = queryset.filter(is_public=True)
# Apply list type filter
if list_type:
queryset = queryset.filter(list_type=list_type)
# Apply user filter
if user_id:
queryset = queryset.filter(user_id=user_id)
# Apply ordering
valid_order_fields = ['created', 'modified', 'title']
order_field = ordering.lstrip('-')
if order_field in valid_order_fields:
queryset = queryset.order_by(ordering)
else:
queryset = queryset.order_by('-created')
# Serialize lists
lists = [_serialize_top_list(tl) for tl in queryset]
return lists
@router.get("/public", response={200: List[TopListOut]})
@paginate(TopListPagination)
def list_public_top_lists(
request,
list_type: Optional[str] = Query(None),
user_id: Optional[UUID] = Query(None),
ordering: Optional[str] = Query("-created")
):
"""
List public top lists.
**Authentication:** Optional
**Parameters:**
- list_type: Filter by type (optional)
- user_id: Filter by user (optional)
- ordering: Sort field (default: -created)
**Returns:** Paginated list of public top lists
"""
queryset = UserTopList.objects.filter(is_public=True).select_related('user')
if list_type:
queryset = queryset.filter(list_type=list_type)
if user_id:
queryset = queryset.filter(user_id=user_id)
valid_order_fields = ['created', 'modified', 'title']
order_field = ordering.lstrip('-')
if order_field in valid_order_fields:
queryset = queryset.order_by(ordering)
else:
queryset = queryset.order_by('-created')
lists = [_serialize_top_list(tl) for tl in queryset]
return lists
@router.get("/{list_id}", response={200: TopListDetailOut, 403: ErrorResponse, 404: ErrorResponse})
def get_top_list(request, list_id: UUID):
"""
Get a specific top list with all items.
**Authentication:** Optional
**Parameters:**
- list_id: List UUID
**Returns:** Top list with all items
**Note:** Private lists only accessible to owner.
"""
user = request.auth if hasattr(request, 'auth') else None
top_list = get_object_or_404(
UserTopList.objects.select_related('user'),
id=list_id
)
# Check access
if not top_list.is_public:
if not user or top_list.user != user:
return 403, {'detail': 'This list is private'}
list_data = _serialize_top_list(top_list, include_items=True)
return 200, list_data
@router.put("/{list_id}", response={200: TopListOut, 403: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def update_top_list(request, list_id: UUID, data: TopListUpdateSchema):
"""
Update a top list.
**Authentication:** Required (must be list owner)
**Parameters:**
- list_id: List UUID
- data: Fields to update
**Returns:** Updated list
"""
user = request.auth
top_list = get_object_or_404(UserTopList, id=list_id)
# Check ownership
if top_list.user != user:
return 403, {'detail': 'You can only update your own lists'}
# Update fields
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(top_list, key, value)
top_list.save()
logger.info(f"Top list updated: {top_list.id} by {user.email}")
list_data = _serialize_top_list(top_list)
return 200, list_data
@router.delete("/{list_id}", response={204: None, 403: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def delete_top_list(request, list_id: UUID):
"""
Delete a top list.
**Authentication:** Required (must be list owner)
**Parameters:**
- list_id: List UUID
**Returns:** No content (204)
**Note:** This also deletes all items in the list.
"""
user = request.auth
top_list = get_object_or_404(UserTopList, id=list_id)
# Check ownership
if top_list.user != user:
return 403, {'detail': 'You can only delete your own lists'}
logger.info(f"Top list deleted: {top_list.id} by {user.email}")
top_list.delete()
return 204, None
# ============================================================================
# List Item Endpoints
# ============================================================================
@router.post("/{list_id}/items", response={201: TopListItemOut, 400: ErrorResponse, 403: ErrorResponse}, auth=jwt_auth)
@require_auth
def add_list_item(request, list_id: UUID, data: TopListItemCreateSchema):
"""
Add an item to a top list.
**Authentication:** Required (must be list owner)
**Parameters:**
- list_id: List UUID
- entity_type: "park" or "ride"
- entity_id: Entity UUID
- position: Position in list (optional, auto-assigned if not provided)
- notes: Notes about this item (optional)
**Returns:** Created list item
"""
try:
user = request.auth
top_list = get_object_or_404(UserTopList, id=list_id)
# Check ownership
if top_list.user != user:
return 403, {'detail': 'You can only modify your own lists'}
# Validate entity
entity, content_type = _get_entity(data.entity_type, data.entity_id)
# Validate entity type matches list type
if top_list.list_type == 'parks' and data.entity_type != 'park':
return 400, {'detail': 'Can only add parks to a parks list'}
elif top_list.list_type in ['rides', 'coasters']:
if data.entity_type != 'ride':
return 400, {'detail': f'Can only add rides to a {top_list.list_type} list'}
if top_list.list_type == 'coasters' and not entity.is_coaster:
return 400, {'detail': 'Can only add coasters to a coasters list'}
# Determine position
if data.position is None:
# Auto-assign position (append to end)
max_pos = top_list.items.aggregate(max_pos=Max('position'))['max_pos']
position = (max_pos or 0) + 1
else:
position = data.position
# Check if position is taken
if top_list.items.filter(position=position).exists():
return 400, {'detail': f'Position {position} is already taken'}
# Create item
with transaction.atomic():
item = UserTopListItem.objects.create(
top_list=top_list,
content_type=content_type,
object_id=entity.id,
position=position,
notes=data.notes or '',
)
logger.info(f"List item added: {item.id} to list {list_id}")
item_data = _serialize_list_item(item)
return 201, item_data
except ValidationError as e:
return 400, {'detail': str(e)}
except Exception as e:
logger.error(f"Error adding list item: {e}")
return 400, {'detail': str(e)}
@router.put("/{list_id}/items/{position}", response={200: TopListItemOut, 400: ErrorResponse, 403: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def update_list_item(request, list_id: UUID, position: int, data: TopListItemUpdateSchema):
"""
Update a list item.
**Authentication:** Required (must be list owner)
**Parameters:**
- list_id: List UUID
- position: Current position
- data: Fields to update (position, notes)
**Returns:** Updated item
**Note:** If changing position, items are reordered automatically.
"""
try:
user = request.auth
top_list = get_object_or_404(UserTopList, id=list_id)
# Check ownership
if top_list.user != user:
return 403, {'detail': 'You can only modify your own lists'}
# Get item
item = get_object_or_404(
UserTopListItem.objects.select_related('content_type'),
top_list=top_list,
position=position
)
with transaction.atomic():
# Handle position change
if data.position is not None and data.position != position:
new_position = data.position
# Check if new position exists
target_item = top_list.items.filter(position=new_position).first()
if target_item:
# Swap positions
target_item.position = position
target_item.save()
item.position = new_position
# Update notes if provided
if data.notes is not None:
item.notes = data.notes
item.save()
logger.info(f"List item updated: {item.id}")
item_data = _serialize_list_item(item)
return 200, item_data
except Exception as e:
logger.error(f"Error updating list item: {e}")
return 400, {'detail': str(e)}
@router.delete("/{list_id}/items/{position}", response={204: None, 403: ErrorResponse, 404: ErrorResponse}, auth=jwt_auth)
@require_auth
def delete_list_item(request, list_id: UUID, position: int):
"""
Remove an item from a list.
**Authentication:** Required (must be list owner)
**Parameters:**
- list_id: List UUID
- position: Position of item to remove
**Returns:** No content (204)
**Note:** Remaining items are automatically reordered.
"""
user = request.auth
top_list = get_object_or_404(UserTopList, id=list_id)
# Check ownership
if top_list.user != user:
return 403, {'detail': 'You can only modify your own lists'}
# Get item
item = get_object_or_404(
UserTopListItem,
top_list=top_list,
position=position
)
with transaction.atomic():
# Delete item
item.delete()
# Reorder remaining items
items_to_reorder = top_list.items.filter(position__gt=position).order_by('position')
for i, remaining_item in enumerate(items_to_reorder, start=position):
remaining_item.position = i
remaining_item.save()
logger.info(f"List item deleted from list {list_id} at position {position}")
return 204, None
# ============================================================================
# User-Specific Endpoints
# ============================================================================
@router.get("/users/{user_id}", response={200: List[TopListOut], 403: ErrorResponse})
@paginate(TopListPagination)
def get_user_top_lists(
request,
user_id: UUID,
list_type: Optional[str] = Query(None),
ordering: Optional[str] = Query("-created")
):
"""
Get a user's top lists.
**Authentication:** Optional
**Parameters:**
- user_id: User UUID
- list_type: Filter by type (optional)
- ordering: Sort field (default: -created)
**Returns:** Paginated list of user's top lists
**Note:** Only public lists visible unless viewing own lists.
"""
target_user = get_object_or_404(User, id=user_id)
# Check if current user
current_user = request.auth if hasattr(request, 'auth') else None
is_owner = current_user and current_user.id == target_user.id
# Build query
queryset = UserTopList.objects.filter(user=target_user).select_related('user')
# Apply visibility filter
if not is_owner:
queryset = queryset.filter(is_public=True)
# Apply list type filter
if list_type:
queryset = queryset.filter(list_type=list_type)
# Apply ordering
valid_order_fields = ['created', 'modified', 'title']
order_field = ordering.lstrip('-')
if order_field in valid_order_fields:
queryset = queryset.order_by(ordering)
else:
queryset = queryset.order_by('-created')
# Serialize lists
lists = [_serialize_top_list(tl) for tl in queryset]
return lists

254
django/api/v1/schemas.py
View File

@@ -967,3 +967,257 @@ class RideSearchFilters(BaseModel):
min_speed: Optional[Decimal] = Field(None, description="Minimum speed in mph")
max_speed: Optional[Decimal] = Field(None, description="Maximum speed in mph")
limit: int = Field(20, ge=1, le=100)
# ============================================================================
# Review Schemas
# ============================================================================
class UserSchema(BaseModel):
"""Minimal user schema for embedding in other schemas."""
id: UUID
username: str
display_name: str
avatar_url: Optional[str] = None
reputation_score: int
class Config:
from_attributes = True
class ReviewCreateSchema(BaseModel):
"""Schema for creating a review."""
entity_type: str = Field(..., description="Entity type: 'park' or 'ride'")
entity_id: UUID = Field(..., description="ID of park or ride being reviewed")
title: str = Field(..., min_length=1, max_length=200, description="Review title")
content: str = Field(..., min_length=10, description="Review content (min 10 characters)")
rating: int = Field(..., ge=1, le=5, description="Rating from 1 to 5 stars")
visit_date: Optional[date] = Field(None, description="Date of visit")
wait_time_minutes: Optional[int] = Field(None, ge=0, description="Wait time in minutes")
@field_validator('entity_type')
def validate_entity_type(cls, v):
if v not in ['park', 'ride']:
raise ValueError('entity_type must be "park" or "ride"')
return v
class ReviewUpdateSchema(BaseModel):
"""Schema for updating a review."""
title: Optional[str] = Field(None, min_length=1, max_length=200)
content: Optional[str] = Field(None, min_length=10)
rating: Optional[int] = Field(None, ge=1, le=5)
visit_date: Optional[date] = None
wait_time_minutes: Optional[int] = Field(None, ge=0)
class VoteRequest(BaseModel):
"""Schema for voting on a review."""
is_helpful: bool = Field(..., description="True if helpful, False if not helpful")
class ReviewOut(TimestampSchema):
"""Schema for review output."""
id: int
user: UserSchema
entity_type: str
entity_id: UUID
entity_name: str
title: str
content: str
rating: int
visit_date: Optional[date]
wait_time_minutes: Optional[int]
helpful_votes: int
total_votes: int
helpful_percentage: Optional[float]
moderation_status: str
moderated_at: Optional[datetime]
moderated_by_email: Optional[str]
photo_count: int
user_vote: Optional[bool] = None # Current user's vote if authenticated
class Config:
from_attributes = True
class VoteResponse(BaseModel):
"""Response for vote action."""
success: bool
review_id: int
helpful_votes: int
total_votes: int
helpful_percentage: Optional[float]
class ReviewListOut(BaseModel):
"""Paginated review list response."""
items: List[ReviewOut]
total: int
page: int
page_size: int
total_pages: int
class ReviewStatsOut(BaseModel):
"""Statistics about reviews for an entity."""
average_rating: float
total_reviews: int
rating_distribution: dict # {1: count, 2: count, 3: count, 4: count, 5: count}
# ============================================================================
# Ride Credit Schemas
# ============================================================================
class RideCreditCreateSchema(BaseModel):
"""Schema for creating a ride credit."""
ride_id: UUID = Field(..., description="ID of ride")
first_ride_date: Optional[date] = Field(None, description="Date of first ride")
ride_count: int = Field(1, ge=1, description="Number of times ridden")
notes: Optional[str] = Field(None, max_length=500, description="Notes about the ride")
class RideCreditUpdateSchema(BaseModel):
"""Schema for updating a ride credit."""
first_ride_date: Optional[date] = None
ride_count: Optional[int] = Field(None, ge=1)
notes: Optional[str] = Field(None, max_length=500)
class RideCreditOut(TimestampSchema):
"""Schema for ride credit output."""
id: UUID
user: UserSchema
ride_id: UUID
ride_name: str
ride_slug: str
park_id: UUID
park_name: str
park_slug: str
is_coaster: bool
first_ride_date: Optional[date]
ride_count: int
notes: str
class Config:
from_attributes = True
class RideCreditListOut(BaseModel):
"""Paginated ride credit list response."""
items: List[RideCreditOut]
total: int
page: int
page_size: int
total_pages: int
class RideCreditStatsOut(BaseModel):
"""Statistics about user's ride credits."""
total_rides: int # Sum of all ride_counts
total_credits: int # Count of unique rides
unique_parks: int
coaster_count: int
first_credit_date: Optional[date]
last_credit_date: Optional[date]
top_park: Optional[str]
top_park_count: int
recent_credits: List[RideCreditOut]
# ============================================================================
# Top List Schemas
# ============================================================================
class TopListCreateSchema(BaseModel):
"""Schema for creating a top list."""
list_type: str = Field(..., description="List type: 'parks', 'rides', or 'coasters'")
title: str = Field(..., min_length=1, max_length=200, description="List title")
description: Optional[str] = Field(None, max_length=1000, description="List description")
is_public: bool = Field(True, description="Whether list is publicly visible")
@field_validator('list_type')
def validate_list_type(cls, v):
if v not in ['parks', 'rides', 'coasters']:
raise ValueError('list_type must be "parks", "rides", or "coasters"')
return v
class TopListUpdateSchema(BaseModel):
"""Schema for updating a top list."""
title: Optional[str] = Field(None, min_length=1, max_length=200)
description: Optional[str] = Field(None, max_length=1000)
is_public: Optional[bool] = None
class TopListItemCreateSchema(BaseModel):
"""Schema for creating a top list item."""
entity_type: str = Field(..., description="Entity type: 'park' or 'ride'")
entity_id: UUID = Field(..., description="ID of park or ride")
position: Optional[int] = Field(None, ge=1, description="Position in list (1 = top)")
notes: Optional[str] = Field(None, max_length=500, description="Notes about this item")
@field_validator('entity_type')
def validate_entity_type(cls, v):
if v not in ['park', 'ride']:
raise ValueError('entity_type must be "park" or "ride"')
return v
class TopListItemUpdateSchema(BaseModel):
"""Schema for updating a top list item."""
position: Optional[int] = Field(None, ge=1, description="New position in list")
notes: Optional[str] = Field(None, max_length=500)
class TopListItemOut(TimestampSchema):
"""Schema for top list item output."""
id: UUID
position: int
entity_type: str
entity_id: UUID
entity_name: str
entity_slug: str
entity_image_url: Optional[str]
park_name: Optional[str] # For rides, show which park
notes: str
class Config:
from_attributes = True
class TopListOut(TimestampSchema):
"""Schema for top list output."""
id: UUID
user: UserSchema
list_type: str
title: str
description: str
is_public: bool
item_count: int
class Config:
from_attributes = True
class TopListDetailOut(TopListOut):
"""Detailed top list with items."""
items: List[TopListItemOut]
class Config:
from_attributes = True
class TopListListOut(BaseModel):
"""Paginated top list response."""
items: List[TopListOut]
total: int
page: int
page_size: int
total_pages: int
class ReorderItemsRequest(BaseModel):
"""Schema for reordering list items."""
item_positions: dict = Field(..., description="Map of item_id to new_position")