feat: ticket purchase flow, QR display, and admin check-in
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled

Ticket purchase:
- tickets blueprint with routes for my tickets list, ticket detail with QR
- Buy tickets form on entry detail page (HTMX-powered)
- Ticket services: create, query, availability checking

Admin check-in:
- ticket_admin blueprint with dashboard, lookup, and check-in routes
- QR scanner/lookup interface with real-time search
- Per-entry ticket list view
- Check-in transitions ticket state to checked_in

Internal API:
- GET /internal/events/tickets endpoint for cross-app queries
- POST /internal/events/tickets/<code>/checkin for programmatic check-in

Template fixes:
- All templates updated: blog.post.calendars.* → calendars.*
- Removed slug=post.slug parameters (standalone events service)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
giles
2026-02-10 00:00:35 +00:00
parent 59a69ed320
commit 1bab546dfc
63 changed files with 1421 additions and 125 deletions

View File

@@ -160,10 +160,13 @@ def register():
@bp.context_processor
async def inject_root():
from ..tickets.services.tickets import get_available_ticket_count
view_args = getattr(request, "view_args", {}) or {}
entry_id = view_args.get("entry_id")
calendar_entry = None
entry_posts = []
ticket_remaining = None
stmt = (
select(CalendarEntry)
@@ -185,10 +188,13 @@ def register():
await g.s.refresh(calendar_entry, ['slot'])
# Fetch associated posts
entry_posts = await get_entry_posts(g.s, calendar_entry.id)
# Get ticket availability
ticket_remaining = await get_available_ticket_count(g.s, calendar_entry.id)
return {
"entry": calendar_entry,
"entry_posts": entry_posts,
"ticket_remaining": ticket_remaining,
}
@bp.get("/")
@require_admin

View File

166
bp/ticket_admin/routes.py Normal file
View File

@@ -0,0 +1,166 @@
"""
Ticket admin blueprint — check-in interface and ticket management.
Routes:
GET /admin/tickets/ — Ticket dashboard (scan + list)
GET /admin/tickets/entry/<id>/ — Tickets for a specific entry
POST /admin/tickets/<code>/checkin — Check in a ticket
GET /admin/tickets/<code>/ — Ticket admin detail
"""
from __future__ import annotations
import logging
from quart import (
Blueprint, g, request, render_template, make_response, jsonify,
)
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
from models.calendars import CalendarEntry, Ticket, TicketType
from suma_browser.app.authz import require_admin
from suma_browser.app.redis_cacher import clear_cache
from ..tickets.services.tickets import (
get_ticket_by_code,
get_tickets_for_entry,
checkin_ticket,
)
logger = logging.getLogger(__name__)
def register() -> Blueprint:
bp = Blueprint("ticket_admin", __name__, url_prefix="/admin/tickets")
@bp.get("/")
@require_admin
async def dashboard():
"""Ticket admin dashboard with QR scanner and recent tickets."""
from suma_browser.app.utils.htmx import is_htmx_request
# Get recent tickets
result = await g.s.execute(
select(Ticket)
.options(
selectinload(Ticket.entry).selectinload(CalendarEntry.calendar),
selectinload(Ticket.ticket_type),
)
.order_by(Ticket.created_at.desc())
.limit(50)
)
tickets = result.scalars().all()
# Stats
total = await g.s.scalar(select(func.count(Ticket.id)))
confirmed = await g.s.scalar(
select(func.count(Ticket.id)).where(Ticket.state == "confirmed")
)
checked_in = await g.s.scalar(
select(func.count(Ticket.id)).where(Ticket.state == "checked_in")
)
reserved = await g.s.scalar(
select(func.count(Ticket.id)).where(Ticket.state == "reserved")
)
stats = {
"total": total or 0,
"confirmed": confirmed or 0,
"checked_in": checked_in or 0,
"reserved": reserved or 0,
}
if not is_htmx_request():
html = await render_template(
"_types/ticket_admin/index.html",
tickets=tickets,
stats=stats,
)
else:
html = await render_template(
"_types/ticket_admin/_main_panel.html",
tickets=tickets,
stats=stats,
)
return await make_response(html, 200)
@bp.get("/entry/<int:entry_id>/")
@require_admin
async def entry_tickets(entry_id: int):
"""List all tickets for a specific calendar entry."""
from suma_browser.app.utils.htmx import is_htmx_request
entry = await g.s.scalar(
select(CalendarEntry)
.where(
CalendarEntry.id == entry_id,
CalendarEntry.deleted_at.is_(None),
)
.options(selectinload(CalendarEntry.calendar))
)
if not entry:
return await make_response("Entry not found", 404)
tickets = await get_tickets_for_entry(g.s, entry_id)
html = await render_template(
"_types/ticket_admin/_entry_tickets.html",
entry=entry,
tickets=tickets,
)
return await make_response(html, 200)
@bp.get("/lookup/")
@require_admin
async def lookup():
"""Look up a ticket by code (used by scanner)."""
code = request.args.get("code", "").strip()
if not code:
return await make_response(
'<div class="text-sm text-stone-500">Enter a ticket code</div>',
200,
)
ticket = await get_ticket_by_code(g.s, code)
if not ticket:
html = await render_template(
"_types/ticket_admin/_lookup_result.html",
ticket=None,
error="Ticket not found",
)
return await make_response(html, 200)
html = await render_template(
"_types/ticket_admin/_lookup_result.html",
ticket=ticket,
error=None,
)
return await make_response(html, 200)
@bp.post("/<code>/checkin/")
@require_admin
@clear_cache(tag="calendars", tag_scope="all")
async def do_checkin(code: str):
"""Check in a ticket by its code."""
success, error = await checkin_ticket(g.s, code)
if not success:
html = await render_template(
"_types/ticket_admin/_checkin_result.html",
success=False,
error=error,
ticket=None,
)
return await make_response(html, 200)
ticket = await get_ticket_by_code(g.s, code)
html = await render_template(
"_types/ticket_admin/_checkin_result.html",
success=True,
error=None,
ticket=ticket,
)
return await make_response(html, 200)
return bp

View File

0
bp/tickets/__init__.py Normal file
View File

181
bp/tickets/routes.py Normal file
View File

@@ -0,0 +1,181 @@
"""
Tickets blueprint — user-facing ticket views and QR codes.
Routes:
GET /tickets/ — My tickets list
GET /tickets/<code>/ — Ticket detail with QR code
POST /tickets/buy/ — Purchase tickets for an entry
"""
from __future__ import annotations
import logging
from quart import (
Blueprint, g, request, render_template, make_response,
)
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from models.calendars import CalendarEntry
from shared.cart_identity import current_cart_identity
from suma_browser.app.redis_cacher import clear_cache
from .services.tickets import (
create_ticket,
get_ticket_by_code,
get_user_tickets,
get_available_ticket_count,
get_tickets_for_entry,
)
logger = logging.getLogger(__name__)
def register() -> Blueprint:
bp = Blueprint("tickets", __name__, url_prefix="/tickets")
@bp.get("/")
async def my_tickets():
"""List all tickets for the current user/session."""
from suma_browser.app.utils.htmx import is_htmx_request
ident = current_cart_identity()
tickets = await get_user_tickets(
g.s,
user_id=ident["user_id"],
session_id=ident["session_id"],
)
if not is_htmx_request():
html = await render_template(
"_types/tickets/index.html",
tickets=tickets,
)
else:
html = await render_template(
"_types/tickets/_main_panel.html",
tickets=tickets,
)
return await make_response(html, 200)
@bp.get("/<code>/")
async def ticket_detail(code: str):
"""View a single ticket with QR code."""
from suma_browser.app.utils.htmx import is_htmx_request
ticket = await get_ticket_by_code(g.s, code)
if not ticket:
return await make_response("Ticket not found", 404)
# Verify ownership
ident = current_cart_identity()
if ident["user_id"] is not None:
if ticket.user_id != ident["user_id"]:
return await make_response("Ticket not found", 404)
elif ident["session_id"] is not None:
if ticket.session_id != ident["session_id"]:
return await make_response("Ticket not found", 404)
else:
return await make_response("Ticket not found", 404)
if not is_htmx_request():
html = await render_template(
"_types/tickets/detail.html",
ticket=ticket,
)
else:
html = await render_template(
"_types/tickets/_detail_panel.html",
ticket=ticket,
)
return await make_response(html, 200)
@bp.post("/buy/")
@clear_cache(tag="calendars", tag_scope="all")
async def buy_tickets():
"""
Purchase tickets for a calendar entry.
Creates ticket records with state='reserved' (awaiting payment).
Form fields:
entry_id — the calendar entry ID
ticket_type_id (optional) — specific ticket type
quantity — number of tickets (default 1)
"""
form = await request.form
entry_id_raw = form.get("entry_id", "").strip()
if not entry_id_raw:
return await make_response("Entry ID required", 400)
try:
entry_id = int(entry_id_raw)
except ValueError:
return await make_response("Invalid entry ID", 400)
# Load entry
entry = await g.s.scalar(
select(CalendarEntry)
.where(
CalendarEntry.id == entry_id,
CalendarEntry.deleted_at.is_(None),
)
.options(selectinload(CalendarEntry.ticket_types))
)
if not entry:
return await make_response("Entry not found", 404)
if entry.ticket_price is None:
return await make_response("Tickets not available for this entry", 400)
# Check availability
available = await get_available_ticket_count(g.s, entry_id)
quantity = int(form.get("quantity", 1))
if quantity < 1:
quantity = 1
if available is not None and quantity > available:
return await make_response(
f"Only {available} ticket(s) remaining", 400
)
# Ticket type (optional)
ticket_type_id = None
tt_raw = form.get("ticket_type_id", "").strip()
if tt_raw:
try:
ticket_type_id = int(tt_raw)
except ValueError:
pass
ident = current_cart_identity()
# Create tickets
created = []
for _ in range(quantity):
ticket = await create_ticket(
g.s,
entry_id=entry_id,
ticket_type_id=ticket_type_id,
user_id=ident["user_id"],
session_id=ident["session_id"],
state="reserved",
)
created.append(ticket)
# Re-check availability for display
remaining = await get_available_ticket_count(g.s, entry_id)
all_tickets = await get_tickets_for_entry(g.s, entry_id)
html = await render_template(
"_types/tickets/_buy_result.html",
entry=entry,
created_tickets=created,
remaining=remaining,
all_tickets=all_tickets,
)
return await make_response(html, 200)
return bp

View File

View File

@@ -0,0 +1,238 @@
"""
Ticket service layer — create, query, and manage tickets.
"""
from __future__ import annotations
import uuid
from decimal import Decimal
from typing import Optional
from sqlalchemy import select, update, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from models.calendars import Ticket, TicketType, CalendarEntry
async def create_ticket(
session: AsyncSession,
*,
entry_id: int,
ticket_type_id: Optional[int] = None,
user_id: Optional[int] = None,
session_id: Optional[str] = None,
order_id: Optional[int] = None,
state: str = "reserved",
) -> Ticket:
"""Create a single ticket with a unique code."""
ticket = Ticket(
entry_id=entry_id,
ticket_type_id=ticket_type_id,
user_id=user_id,
session_id=session_id,
order_id=order_id,
code=uuid.uuid4().hex,
state=state,
)
session.add(ticket)
await session.flush()
return ticket
async def create_tickets_for_order(
session: AsyncSession,
order_id: int,
user_id: Optional[int],
session_id: Optional[str],
) -> list[Ticket]:
"""
Create ticket records for all calendar entries in an order
that have ticket_price configured.
Called during checkout after calendar entries are transitioned to 'ordered'.
"""
# Find all ordered entries for this order that have ticket pricing
result = await session.execute(
select(CalendarEntry)
.where(
CalendarEntry.order_id == order_id,
CalendarEntry.deleted_at.is_(None),
CalendarEntry.ticket_price.isnot(None),
)
.options(selectinload(CalendarEntry.ticket_types))
)
entries = result.scalars().all()
tickets = []
for entry in entries:
if entry.ticket_types:
# Entry has specific ticket types — create one ticket per type
# (quantity handling can be added later)
for tt in entry.ticket_types:
if tt.deleted_at is None:
ticket = await create_ticket(
session,
entry_id=entry.id,
ticket_type_id=tt.id,
user_id=user_id,
session_id=session_id,
order_id=order_id,
state="reserved",
)
tickets.append(ticket)
else:
# Simple ticket — one per entry
ticket = await create_ticket(
session,
entry_id=entry.id,
user_id=user_id,
session_id=session_id,
order_id=order_id,
state="reserved",
)
tickets.append(ticket)
return tickets
async def confirm_tickets_for_order(
session: AsyncSession,
order_id: int,
) -> int:
"""
Transition tickets from reserved → confirmed when payment succeeds.
Returns the number of tickets confirmed.
"""
result = await session.execute(
update(Ticket)
.where(
Ticket.order_id == order_id,
Ticket.state == "reserved",
)
.values(state="confirmed")
)
return result.rowcount
async def get_ticket_by_code(
session: AsyncSession,
code: str,
) -> Optional[Ticket]:
"""Look up a ticket by its unique code."""
result = await session.execute(
select(Ticket)
.where(Ticket.code == code)
.options(
selectinload(Ticket.entry).selectinload(CalendarEntry.calendar),
selectinload(Ticket.ticket_type),
)
)
return result.scalar_one_or_none()
async def get_user_tickets(
session: AsyncSession,
user_id: Optional[int] = None,
session_id: Optional[str] = None,
state: Optional[str] = None,
) -> list[Ticket]:
"""Get all tickets for a user or session."""
filters = []
if user_id is not None:
filters.append(Ticket.user_id == user_id)
elif session_id is not None:
filters.append(Ticket.session_id == session_id)
else:
return []
if state:
filters.append(Ticket.state == state)
else:
# Exclude cancelled by default
filters.append(Ticket.state != "cancelled")
result = await session.execute(
select(Ticket)
.where(*filters)
.options(
selectinload(Ticket.entry).selectinload(CalendarEntry.calendar),
selectinload(Ticket.ticket_type),
)
.order_by(Ticket.created_at.desc())
)
return result.scalars().all()
async def get_tickets_for_entry(
session: AsyncSession,
entry_id: int,
) -> list[Ticket]:
"""Get all non-cancelled tickets for a calendar entry."""
result = await session.execute(
select(Ticket)
.where(
Ticket.entry_id == entry_id,
Ticket.state != "cancelled",
)
.options(
selectinload(Ticket.ticket_type),
)
.order_by(Ticket.created_at.asc())
)
return result.scalars().all()
async def get_available_ticket_count(
session: AsyncSession,
entry_id: int,
) -> Optional[int]:
"""
Get number of remaining tickets for an entry.
Returns None if unlimited.
"""
entry = await session.scalar(
select(CalendarEntry).where(
CalendarEntry.id == entry_id,
CalendarEntry.deleted_at.is_(None),
)
)
if not entry or entry.ticket_price is None:
return None
if entry.ticket_count is None:
return None # Unlimited
# Count non-cancelled tickets
sold = await session.scalar(
select(func.count(Ticket.id)).where(
Ticket.entry_id == entry_id,
Ticket.state != "cancelled",
)
)
return max(0, entry.ticket_count - (sold or 0))
async def checkin_ticket(
session: AsyncSession,
code: str,
) -> tuple[bool, Optional[str]]:
"""
Check in a ticket by its code.
Returns (success, error_message).
"""
from datetime import datetime, timezone
ticket = await get_ticket_by_code(session, code)
if not ticket:
return False, "Ticket not found"
if ticket.state == "checked_in":
return False, "Ticket already checked in"
if ticket.state == "cancelled":
return False, "Ticket is cancelled"
if ticket.state not in ("confirmed", "reserved"):
return False, f"Ticket in unexpected state: {ticket.state}"
ticket.state = "checked_in"
ticket.checked_in_at = datetime.now(timezone.utc)
return True, None