95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from app.models.catalog import Event, EventMedia, TicketType, Venue
|
|
from app.models.enums import EventStatus, EventType
|
|
from app.schemas.catalog import EventDetail, EventSummary, TicketTypePublic, VenuePublic
|
|
|
|
|
|
def _ticket_type_public(ticket: TicketType) -> TicketTypePublic:
|
|
return TicketTypePublic(
|
|
id=ticket.id,
|
|
name=ticket.name,
|
|
description=ticket.description,
|
|
price_paise=ticket.price_paise,
|
|
status=ticket.status,
|
|
sort_order=ticket.sort_order,
|
|
remaining_quantity=max(ticket.quantity_total - ticket.quantity_sold, 0),
|
|
)
|
|
|
|
|
|
def list_events(
|
|
db: Session,
|
|
*,
|
|
category: EventType | None,
|
|
status: EventStatus | None,
|
|
upcoming: bool,
|
|
) -> list[Event]:
|
|
query = (
|
|
select(Event)
|
|
.options(joinedload(Event.venue))
|
|
.order_by(Event.starts_at.asc())
|
|
)
|
|
|
|
if status is not None:
|
|
query = query.where(Event.status == status)
|
|
if category is not None:
|
|
query = query.where(Event.category == category)
|
|
if upcoming:
|
|
now = datetime.now(timezone.utc)
|
|
query = query.where(Event.starts_at >= now)
|
|
|
|
return list(db.scalars(query).unique().all())
|
|
|
|
|
|
def get_event_by_slug(db: Session, slug: str) -> Event:
|
|
event = db.scalar(
|
|
select(Event)
|
|
.options(
|
|
joinedload(Event.venue),
|
|
joinedload(Event.media),
|
|
joinedload(Event.ticket_types),
|
|
)
|
|
.where(Event.slug == slug)
|
|
)
|
|
if not event:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event not found")
|
|
return event
|
|
|
|
|
|
def event_to_summary(event: Event) -> EventSummary:
|
|
return EventSummary.model_validate(event)
|
|
|
|
|
|
def event_to_detail(event: Event) -> EventDetail:
|
|
return EventDetail(
|
|
id=event.id,
|
|
slug=event.slug,
|
|
title=event.title,
|
|
category=event.category,
|
|
status=event.status,
|
|
starts_at=event.starts_at,
|
|
ends_at=event.ends_at,
|
|
is_featured=event.is_featured,
|
|
cover_image_url=event.cover_image_url,
|
|
cover_image_alt=event.cover_image_alt,
|
|
venue=VenuePublic.model_validate(event.venue),
|
|
description=event.description,
|
|
meta_description=event.meta_description,
|
|
media=event.media,
|
|
ticket_types=[_ticket_type_public(ticket) for ticket in event.ticket_types],
|
|
)
|
|
|
|
|
|
def list_venues(db: Session) -> list[Venue]:
|
|
return list(db.scalars(select(Venue).order_by(Venue.name.asc())).all())
|
|
|
|
|
|
def get_venue_by_id(db: Session, venue_id: int) -> Venue:
|
|
venue = db.get(Venue, venue_id)
|
|
if not venue:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Venue not found")
|
|
return venue
|