Amirhossein Khalili
6c5fadef2f
Replace persisted event statuses with publication and cancellation flags, and compute public lifecycle labels from event timing and registration capacity. Keep filters, registration checks, and profile responses aligned with the derived status.
479 lines
16 KiB
Python
479 lines
16 KiB
Python
"""Event and gallery API schemas."""
|
|
|
|
from uuid import UUID
|
|
from ninja import ModelSchema, Schema
|
|
from pydantic import field_validator
|
|
from typing import Literal, Optional, List
|
|
from datetime import datetime
|
|
from django.db.models import Q
|
|
from django.utils import timezone
|
|
|
|
from apps.blog.api.schemas import AuthorSchema
|
|
from apps.events.models import Event, Registration
|
|
from apps.gallery.models import Gallery
|
|
from apps.payments.models import Payment
|
|
from core.media import BLUR_VARIANT, PREVIEW_VARIANT, THUMBNAIL_VARIANT, derivative_url
|
|
|
|
|
|
class EventGallerySchema(ModelSchema):
|
|
"""Schema representing gallery items associated with an event."""
|
|
uploaded_by: AuthorSchema
|
|
file_size_mb: float
|
|
markdown_url: str
|
|
absolute_image_url: Optional[str] = None
|
|
absolute_image_preview_url: Optional[str] = None
|
|
absolute_image_blur_url: Optional[str] = None
|
|
|
|
class Meta:
|
|
model = Gallery
|
|
fields = ['id', 'title', 'description', 'image', 'alt_text',
|
|
'width', 'height', 'is_public', 'created_at']
|
|
|
|
@staticmethod
|
|
def resolve_absolute_image_url(obj, context):
|
|
request = context['request']
|
|
if obj.image and hasattr(obj.image, 'url'):
|
|
return request.build_absolute_uri(obj.image.url)
|
|
return None
|
|
|
|
@staticmethod
|
|
def resolve_absolute_image_preview_url(obj, context):
|
|
request = context["request"]
|
|
url = derivative_url(obj.image, PREVIEW_VARIANT)
|
|
return request.build_absolute_uri(url) if url else None
|
|
|
|
@staticmethod
|
|
def resolve_absolute_image_blur_url(obj, context):
|
|
request = context["request"]
|
|
url = derivative_url(obj.image, BLUR_VARIANT)
|
|
return request.build_absolute_uri(url) if url else None
|
|
|
|
class EventSchema(ModelSchema):
|
|
"""Schema providing full event details for API responses."""
|
|
gallery_images: List[EventGallerySchema]
|
|
description_html: str
|
|
registration_count: int
|
|
absolute_featured_image_url: Optional[str] = None
|
|
absolute_featured_image_thumbnail_url: Optional[str] = None
|
|
absolute_featured_image_preview_url: Optional[str] = None
|
|
status: str
|
|
status_label: str
|
|
effective_registration_end_date: datetime
|
|
|
|
class Meta:
|
|
model = Event
|
|
fields = [
|
|
'id', 'title', 'slug', 'description', 'featured_image', 'event_type',
|
|
'address', 'location', 'online_link', 'start_time', 'end_time',
|
|
'registration_start_date', 'registration_end_date', 'registration_success_markdown',
|
|
'capacity', 'price', 'is_published',
|
|
'is_cancelled', 'created_at', 'updated_at'
|
|
]
|
|
|
|
@staticmethod
|
|
def resolve_absolute_featured_image_url(obj, context):
|
|
request = context['request']
|
|
if obj.featured_image and hasattr(obj.featured_image, 'url'):
|
|
return request.build_absolute_uri(obj.featured_image.url)
|
|
return None
|
|
|
|
@staticmethod
|
|
def resolve_absolute_featured_image_thumbnail_url(obj, context):
|
|
request = context["request"]
|
|
url = derivative_url(obj.featured_image, THUMBNAIL_VARIANT)
|
|
return request.build_absolute_uri(url) if url else None
|
|
|
|
@staticmethod
|
|
def resolve_absolute_featured_image_preview_url(obj, context):
|
|
request = context["request"]
|
|
url = derivative_url(obj.featured_image, PREVIEW_VARIANT)
|
|
return request.build_absolute_uri(url) if url else None
|
|
|
|
@staticmethod
|
|
def resolve_registration_count(obj):
|
|
if hasattr(obj, "browse_registration_count"):
|
|
return obj.browse_registration_count
|
|
now = timezone.now()
|
|
return obj.registrations.filter(is_deleted=False).filter(
|
|
Q(status__in=[Registration.StatusChoices.CONFIRMED, Registration.StatusChoices.ATTENDED])
|
|
| Q(status=Registration.StatusChoices.PENDING, checkout_expires_at__gt=now)
|
|
).count()
|
|
|
|
@staticmethod
|
|
def resolve_description_html(obj):
|
|
return obj.description_html
|
|
|
|
|
|
class EventListSchema(Schema):
|
|
"""Condensed event representation for list endpoints."""
|
|
id: int
|
|
title: str
|
|
slug: str
|
|
featured_image: Optional[str] = None
|
|
absolute_featured_image_url: Optional[str] = None
|
|
absolute_featured_image_thumbnail_url: Optional[str] = None
|
|
absolute_featured_image_preview_url: Optional[str] = None
|
|
event_type: str
|
|
address: Optional[str] = None
|
|
start_time: datetime
|
|
end_time: datetime
|
|
registration_start_date: Optional[datetime] = None
|
|
registration_end_date: Optional[datetime] = None
|
|
effective_registration_end_date: datetime
|
|
capacity: Optional[int] = None
|
|
price: Optional[float] = None
|
|
status: str
|
|
status_label: str
|
|
is_published: bool
|
|
is_cancelled: bool
|
|
registration_count: int
|
|
created_at: datetime
|
|
|
|
@staticmethod
|
|
def resolve_absolute_featured_image_url(obj, context):
|
|
request = context['request']
|
|
if obj.featured_image and hasattr(obj.featured_image, 'url'):
|
|
return request.build_absolute_uri(obj.featured_image.url)
|
|
return None
|
|
|
|
@staticmethod
|
|
def resolve_absolute_featured_image_thumbnail_url(obj, context):
|
|
request = context["request"]
|
|
url = derivative_url(obj.featured_image, THUMBNAIL_VARIANT)
|
|
return request.build_absolute_uri(url) if url else None
|
|
|
|
@staticmethod
|
|
def resolve_absolute_featured_image_preview_url(obj, context):
|
|
request = context["request"]
|
|
url = derivative_url(obj.featured_image, PREVIEW_VARIANT)
|
|
return request.build_absolute_uri(url) if url else None
|
|
|
|
@staticmethod
|
|
def resolve_registration_count(obj):
|
|
if hasattr(obj, "browse_registration_count"):
|
|
return obj.browse_registration_count
|
|
now = timezone.now()
|
|
return obj.registrations.filter(is_deleted=False).filter(
|
|
Q(status__in=[Registration.StatusChoices.CONFIRMED, Registration.StatusChoices.ATTENDED])
|
|
| Q(status=Registration.StatusChoices.PENDING, checkout_expires_at__gt=now)
|
|
).count()
|
|
|
|
|
|
class PaginatedEventBrowseSchema(Schema):
|
|
count: int
|
|
page: int
|
|
page_size: int
|
|
total_pages: int
|
|
price_ceiling: int
|
|
results: List[EventListSchema]
|
|
|
|
class EventCreateSchema(Schema):
|
|
"""Payload for creating events via the API."""
|
|
title: str
|
|
slug: Optional[str] = None
|
|
description: str
|
|
event_type: str
|
|
address: Optional[str] = None
|
|
location: Optional[str] = None
|
|
online_link: Optional[str] = None
|
|
start_time: datetime
|
|
end_time: datetime
|
|
registration_start_date: Optional[datetime] = None
|
|
registration_end_date: Optional[datetime] = None
|
|
capacity: Optional[int] = None
|
|
price: Optional[float] = None
|
|
is_published: bool = False
|
|
is_cancelled: bool = False
|
|
registration_success_markdown: Optional[str] = None
|
|
gallery_image_ids: Optional[List[int]] = []
|
|
|
|
class EventUpdateSchema(Schema):
|
|
"""Payload for updating events via the API."""
|
|
title: Optional[str] = None
|
|
slug: Optional[str] = None
|
|
description: Optional[str] = None
|
|
event_type: Optional[str] = None
|
|
address: Optional[str] = None
|
|
location: Optional[str] = None
|
|
online_link: Optional[str] = None
|
|
start_time: Optional[datetime] = None
|
|
end_time: Optional[datetime] = None
|
|
registration_start_date: Optional[datetime] = None
|
|
registration_end_date: Optional[datetime] = None
|
|
capacity: Optional[int] = None
|
|
price: Optional[float] = None
|
|
is_published: Optional[bool] = None
|
|
is_cancelled: Optional[bool] = None
|
|
registration_success_markdown: Optional[str] = None
|
|
gallery_image_ids: Optional[List[int]] = None
|
|
|
|
class RegistrationSchema(ModelSchema):
|
|
"""Schema describing a registration entry with event context."""
|
|
user: AuthorSchema
|
|
event: EventListSchema
|
|
discount_code: str | None = None
|
|
|
|
class Meta:
|
|
model = Registration
|
|
fields = [
|
|
'id',
|
|
'status',
|
|
'registered_at',
|
|
'checkout_started_at',
|
|
'checkout_expires_at',
|
|
'cancellation_reason',
|
|
'ticket_id',
|
|
'discount_amount',
|
|
'final_price',
|
|
'created_at',
|
|
'updated_at',
|
|
]
|
|
|
|
@staticmethod
|
|
def resolve_discount_code(obj):
|
|
return obj.discount_code.code if obj.discount_code else None
|
|
|
|
|
|
class AdminUserSchema(Schema):
|
|
id: int
|
|
username: str
|
|
first_name: str
|
|
last_name: str
|
|
email: str
|
|
mobile: Optional[str] = None
|
|
profile_picture: Optional[str] = None
|
|
profile_picture_thumbnail_url: Optional[str] = None
|
|
profile_picture_preview_url: Optional[str] = None
|
|
university: Optional[str] = None
|
|
major: Optional[str] = None
|
|
student_id: Optional[str] = None
|
|
year_of_study: Optional[int] = None
|
|
|
|
@staticmethod
|
|
def resolve_profile_picture(obj, context):
|
|
image = getattr(obj, "profile_picture", None)
|
|
if not getattr(image, "name", None):
|
|
return None
|
|
request = context["request"]
|
|
return request.build_absolute_uri(image.url) if hasattr(image, "url") else None
|
|
|
|
@staticmethod
|
|
def resolve_profile_picture_thumbnail_url(obj, context):
|
|
image = getattr(obj, "profile_picture", None)
|
|
if not getattr(image, "name", None):
|
|
return None
|
|
request = context["request"]
|
|
url = derivative_url(image, THUMBNAIL_VARIANT)
|
|
return request.build_absolute_uri(url) if url else None
|
|
|
|
@staticmethod
|
|
def resolve_profile_picture_preview_url(obj, context):
|
|
image = getattr(obj, "profile_picture", None)
|
|
if not getattr(image, "name", None):
|
|
return None
|
|
request = context["request"]
|
|
url = derivative_url(image, PREVIEW_VARIANT)
|
|
return request.build_absolute_uri(url) if url else None
|
|
|
|
@staticmethod
|
|
def resolve_university(obj):
|
|
return obj.get_university_display()
|
|
|
|
@staticmethod
|
|
def resolve_major(obj):
|
|
return obj.get_major_display()
|
|
|
|
|
|
class PaymentAdminSchema(Schema):
|
|
id: int
|
|
authority: Optional[str]
|
|
ref_id: Optional[str]
|
|
card_pan: Optional[str]
|
|
card_hash: Optional[str]
|
|
status: int
|
|
status_label: str
|
|
base_amount: int
|
|
discount_amount: int
|
|
amount: int
|
|
verified_at: Optional[datetime]
|
|
created_at: datetime
|
|
discount_code: Optional[str]
|
|
user: AdminUserSchema
|
|
|
|
@field_validator("discount_code", mode="before")
|
|
def normalize_discount_code(cls, value):
|
|
if value is None:
|
|
return None
|
|
if hasattr(value, "code"):
|
|
return value.code
|
|
return str(value)
|
|
|
|
|
|
class RegistrationAdminSchema(Schema):
|
|
id: int
|
|
ticket_id: UUID
|
|
status: str
|
|
status_label: str
|
|
registered_at: datetime
|
|
final_price: Optional[int]
|
|
discount_amount: Optional[int]
|
|
user: AdminUserSchema
|
|
payments: List[PaymentAdminSchema]
|
|
|
|
|
|
class EventAdminDetailSchema(EventSchema):
|
|
registrations: List[RegistrationAdminSchema] = []
|
|
|
|
@staticmethod
|
|
def resolve_registrations(obj):
|
|
return obj.registrations.select_related("user", "user__university", "user__major").prefetch_related(
|
|
"payments__discount_code"
|
|
).order_by("-registered_at")
|
|
|
|
class PaginatedRegistrationSchema(Schema):
|
|
count: int
|
|
next: Optional[str] = None
|
|
previous: Optional[str] = None
|
|
results: List[RegistrationAdminSchema]
|
|
|
|
class RegistrationStatusUpdateSchema(Schema):
|
|
status: str
|
|
|
|
class RegisterationDetailSchema(Schema):
|
|
"""Detailed registration information with associated event metadata."""
|
|
event_image: Optional[str]
|
|
event_title: str
|
|
event_type: str
|
|
ticket_id: UUID
|
|
status: str
|
|
registered_at: datetime
|
|
success_markdown: Optional[str]
|
|
|
|
class EventBriefSchema(Schema):
|
|
"""Minimal event representation used for nested responses."""
|
|
id: int
|
|
title: str
|
|
slug: str
|
|
start_date: datetime
|
|
end_date: Optional[datetime] = None
|
|
start_time: datetime
|
|
end_time: datetime
|
|
registration_start_date: Optional[datetime] = None
|
|
registration_end_date: Optional[datetime] = None
|
|
effective_registration_end_date: datetime
|
|
capacity: Optional[int] = None
|
|
registration_count: int = 0
|
|
status: str
|
|
status_label: str
|
|
is_published: bool = True
|
|
is_cancelled: bool = False
|
|
location: Optional[str] = None
|
|
price: int
|
|
absolute_image_url: Optional[str] = None
|
|
|
|
class MyEventRegistrationOut(Schema):
|
|
"""Registration information as returned to authenticated users."""
|
|
id: int
|
|
ticket_id: UUID
|
|
created_at: datetime
|
|
status: Literal["pending", "confirmed", "cancelled", "attended"]
|
|
event: EventBriefSchema
|
|
checkout_state: Literal["none", "active", "legacy_pending", "expired", "unresolved", "cancelled"] = "none"
|
|
checkout_expires_at: Optional[datetime] = None
|
|
payment_id: Optional[int] = None
|
|
payment_status: Optional[Literal["initiated", "pending", "paid", "failed", "canceled"]] = None
|
|
refund_status: Optional[Literal["none", "pending", "succeeded", "manual_review"]] = None
|
|
can_resume: bool = False
|
|
|
|
|
|
class PaginatedMyEventRegistrationsSchema(Schema):
|
|
count: int
|
|
page: int
|
|
page_size: int
|
|
total_pages: int
|
|
results: List[MyEventRegistrationOut]
|
|
|
|
class RegistrationStatusOut(Schema):
|
|
is_registered: bool
|
|
checkout_state: Literal[
|
|
"none", "confirmed", "active", "legacy_pending", "expired", "unresolved", "cancelled"
|
|
] = "none"
|
|
registration_id: Optional[int] = None
|
|
registration_status: Optional[str] = None
|
|
ticket_id: Optional[UUID] = None
|
|
checkout_expires_at: Optional[datetime] = None
|
|
payment_id: Optional[int] = None
|
|
payment_status: Optional[Literal["initiated", "pending", "paid", "failed", "canceled"]] = None
|
|
refund_status: Optional[Literal["none", "pending", "succeeded", "manual_review"]] = None
|
|
can_resume: bool = False
|
|
can_cancel: bool = False
|
|
|
|
|
|
class RegistrationHubEventOut(Schema):
|
|
id: int
|
|
title: str
|
|
slug: str
|
|
description: str
|
|
event_type: str
|
|
address: Optional[str] = None
|
|
location: Optional[str] = None
|
|
start_time: datetime
|
|
end_time: datetime
|
|
registration_start_date: Optional[datetime] = None
|
|
registration_end_date: Optional[datetime] = None
|
|
effective_registration_end_date: datetime
|
|
capacity: Optional[int] = None
|
|
price: int
|
|
status: str
|
|
status_label: str
|
|
is_published: bool
|
|
is_cancelled: bool
|
|
registration_count: int
|
|
image_url: Optional[str] = None
|
|
registration_success_markdown: Optional[str] = None
|
|
|
|
|
|
class RegistrationHubRegistrationOut(Schema):
|
|
ticket_id: UUID
|
|
status: str
|
|
registered_at: datetime
|
|
checkout_started_at: Optional[datetime] = None
|
|
checkout_expires_at: Optional[datetime] = None
|
|
cancellation_reason: str = ""
|
|
final_price: Optional[int] = None
|
|
discount_amount: int = 0
|
|
|
|
|
|
class RegistrationHubPaymentOut(Schema):
|
|
id: int
|
|
status: Literal["initiated", "pending", "paid", "failed", "canceled"]
|
|
verification_state: Literal[
|
|
"awaiting_verification", "checking", "retry_scheduled", "manual_review",
|
|
"verified", "failed", "canceled",
|
|
]
|
|
refund_status: Literal["none", "pending", "succeeded", "manual_review"]
|
|
base_amount: int
|
|
discount_amount: int
|
|
amount: int
|
|
ref_id: Optional[str] = None
|
|
created_at: datetime
|
|
gateway_paid_at: Optional[datetime] = None
|
|
|
|
|
|
class RegistrationHubOut(Schema):
|
|
state: Literal[
|
|
"ready", "active_checkout", "legacy_pending", "verification_pending",
|
|
"manual_review", "paid_success", "free_success", "failed", "cancelled",
|
|
"expired", "refund_pending", "refund_manual_review", "refund_succeeded",
|
|
]
|
|
event: RegistrationHubEventOut
|
|
registration: Optional[RegistrationHubRegistrationOut] = None
|
|
payment: Optional[RegistrationHubPaymentOut] = None
|
|
can_register: bool = False
|
|
can_resume: bool = False
|
|
can_cancel: bool = False
|
|
|
|
|
|
class RegistrationCreateSchema(Schema):
|
|
discount_code: Optional[str] = None
|