This repository has been archived on 2026-02-24. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
events/app.py
giles 0a8d1391f6
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 43s
Add glue layer: replace /internal/menu-items API with direct DB query
- Context processor: get_navigation_tree() replaces api_get("coop", "/internal/menu-items")
- Add glue submodule

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-11 23:37:54 +00:00

153 lines
4.6 KiB
Python

from __future__ import annotations
import path_setup # noqa: F401 # adds shared_lib to sys.path
from pathlib import Path
from quart import g, abort
from jinja2 import FileSystemLoader, ChoiceLoader
from sqlalchemy import select
from shared.infrastructure.factory import create_base_app
from bp import register_calendars, register_markets, register_payments
async def events_context() -> dict:
"""
Events app context processor.
- menu_items: direct DB query via glue layer
- cart_count/cart_total: fetched from cart internal API
"""
from shared.infrastructure.context import base_context
from glue.services.navigation import get_navigation_tree
from shared.infrastructure.internal_api import get as api_get
ctx = await base_context()
ctx["menu_items"] = await get_navigation_tree(g.s)
# Cart data from cart API
cart_data = await api_get("cart", "/internal/cart/summary", forward_session=True)
if cart_data:
ctx["cart_count"] = cart_data.get("count", 0)
ctx["cart_total"] = cart_data.get("total", 0)
else:
ctx["cart_count"] = 0
ctx["cart_total"] = 0
return ctx
def create_app() -> "Quart":
from blog.models.ghost_content import Post
from models.calendars import Calendar
from market.models.market_place import MarketPlace
app = create_base_app("events", context_fn=events_context)
# App-specific templates override shared templates
app_templates = str(Path(__file__).resolve().parent / "templates")
app.jinja_loader = ChoiceLoader([
FileSystemLoader(app_templates),
app.jinja_loader,
])
# Calendars nested under post slug: /<slug>/calendars/...
app.register_blueprint(
register_calendars(),
url_prefix="/<slug>/calendars",
)
# Markets nested under post slug: /<slug>/markets/...
app.register_blueprint(
register_markets(),
url_prefix="/<slug>/markets",
)
# Payments nested under post slug: /<slug>/payments/...
app.register_blueprint(
register_payments(),
url_prefix="/<slug>/payments",
)
# --- Auto-inject slug into url_for() calls ---
@app.url_value_preprocessor
def pull_slug(endpoint, values):
if values and "slug" in values:
g.post_slug = values.pop("slug")
@app.url_defaults
def inject_slug(endpoint, values):
slug = g.get("post_slug")
if slug and "slug" not in values:
if app.url_map.is_endpoint_expecting(endpoint, "slug"):
values["slug"] = slug
# --- Load post data for slug ---
@app.before_request
async def hydrate_post():
slug = getattr(g, "post_slug", None)
if not slug:
return
post = (
await g.s.execute(
select(Post).where(Post.slug == slug)
)
).scalar_one_or_none()
if not post:
abort(404)
g.post_data = {
"post": {
"id": post.id,
"title": post.title,
"slug": post.slug,
"feature_image": post.feature_image,
"status": post.status,
"visibility": post.visibility,
},
}
@app.context_processor
async def inject_post():
post_data = getattr(g, "post_data", None)
if not post_data:
return {}
post_id = post_data["post"]["id"]
calendars = (
await g.s.execute(
select(Calendar)
.where(Calendar.container_type == "page", Calendar.container_id == post_id, Calendar.deleted_at.is_(None))
.order_by(Calendar.name.asc())
)
).scalars().all()
markets = (
await g.s.execute(
select(MarketPlace)
.where(MarketPlace.container_type == "page", MarketPlace.container_id == post_id, MarketPlace.deleted_at.is_(None))
.order_by(MarketPlace.name.asc())
)
).scalars().all()
return {
**post_data,
"calendars": calendars,
"markets": markets,
}
# Tickets blueprint — user-facing ticket views and QR codes
from bp.tickets.routes import register as register_tickets
app.register_blueprint(register_tickets())
# Ticket admin — check-in interface (admin only)
from bp.ticket_admin.routes import register as register_ticket_admin
app.register_blueprint(register_ticket_admin())
# Internal API (server-to-server, CSRF-exempt)
from events_api import register as register_events_api
app.register_blueprint(register_events_api())
return app
app = create_app()