from __future__ import annotations import path_setup # noqa: F401 # adds shared/ to sys.path from pathlib import Path from quart import g, abort, request from jinja2 import FileSystemLoader, ChoiceLoader from shared.infrastructure.factory import create_base_app from bp import register_all_events, register_calendars, register_markets, register_payments, register_page, register_fragments, register_actions, register_data async def events_context() -> dict: """ Events app context processor. - nav_tree_html: fetched from blog as fragment - cart_count/cart_total: via cart service (shared DB) """ from shared.infrastructure.context import base_context from shared.infrastructure.cart_identity import current_cart_identity from shared.infrastructure.fragments import fetch_fragments from shared.infrastructure.data_client import fetch_data from shared.contracts.dtos import CartSummaryDTO, dto_from_dict ctx = await base_context() # menu_nodes lives in db_blog; nav-tree fragment provides the real nav ctx["menu_items"] = [] # Cart data via internal data endpoint ident = current_cart_identity() summary_params = {} if ident["user_id"] is not None: summary_params["user_id"] = ident["user_id"] if ident["session_id"] is not None: summary_params["session_id"] = ident["session_id"] raw = await fetch_data("cart", "cart-summary", params=summary_params, required=False) summary = dto_from_dict(CartSummaryDTO, raw) if raw else CartSummaryDTO() ctx["cart_count"] = summary.count + summary.calendar_count + summary.ticket_count ctx["cart_total"] = float(summary.total + summary.calendar_total + summary.ticket_total) # Pre-fetch cross-app HTML fragments concurrently user = getattr(g, "user", None) cart_params = {} if ident["user_id"] is not None: cart_params["user_id"] = ident["user_id"] if ident["session_id"] is not None: cart_params["session_id"] = ident["session_id"] cart_mini_html, auth_menu_html, nav_tree_html = await fetch_fragments([ ("cart", "cart-mini", cart_params or None), ("account", "auth-menu", {"email": user.email} if user else None), ("blog", "nav-tree", {"app_name": "events", "path": request.path}), ]) ctx["cart_mini_html"] = cart_mini_html ctx["auth_menu_html"] = auth_menu_html ctx["nav_tree_html"] = nav_tree_html return ctx def create_app() -> "Quart": from services import register_domain_services app = create_base_app( "events", context_fn=events_context, domain_services_fn=register_domain_services, ) # App-specific templates override shared templates app_templates = str(Path(__file__).resolve().parent / "templates") app.jinja_loader = ChoiceLoader([ FileSystemLoader(app_templates), app.jinja_loader, ]) # All events: / — global view across all pages app.register_blueprint( register_all_events(), url_prefix="/", ) # Page summary: // — upcoming events across all calendars app.register_blueprint( register_page(), url_prefix="/", ) # Calendars nested under post slug: //calendars/... app.register_blueprint( register_calendars(), url_prefix="//calendars", ) # Markets nested under post slug: //markets/... app.register_blueprint( register_markets(), url_prefix="//markets", ) # Payments nested under post slug: //payments/... app.register_blueprint( register_payments(), url_prefix="//payments", ) app.register_blueprint(register_fragments()) app.register_blueprint(register_actions()) app.register_blueprint(register_data()) # --- 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(): from shared.infrastructure.data_client import fetch_data slug = getattr(g, "post_slug", None) if not slug: return post = await fetch_data("blog", "post-by-slug", params={"slug": slug}) if not post: abort(404) g.post_data = { "post": { "id": post["id"], "title": post["title"], "slug": post["slug"], "feature_image": post.get("feature_image"), "status": post["status"], "visibility": post["visibility"], }, } @app.context_processor async def inject_post(): from shared.infrastructure.data_client import fetch_data from shared.contracts.dtos import CalendarDTO, MarketPlaceDTO, dto_from_dict from shared.services.registry import services post_data = getattr(g, "post_data", None) if not post_data: return {} post_id = post_data["post"]["id"] # Calendar data is local (events owns it) calendars = await services.calendar.calendars_for_container(g.s, "page", post_id) # Market data is cross-app raw_markets = await fetch_data("market", "marketplaces-for-container", params={"type": "page", "id": post_id}, required=False) or [] markets = [dto_from_dict(MarketPlaceDTO, m) for m in raw_markets] 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()) # --- oEmbed endpoint --- @app.get("/oembed") async def oembed(): from urllib.parse import urlparse from quart import jsonify from shared.infrastructure.urls import events_url from shared.infrastructure.oembed import build_oembed_response from shared.infrastructure.data_client import fetch_data url = request.args.get("url", "") if not url: return jsonify({"error": "url parameter required"}), 400 parsed = urlparse(url) slug = parsed.path.strip("/").split("/")[0] if parsed.path.strip("/") else "" if not slug: return jsonify({"error": "could not extract slug"}), 404 post = await fetch_data("blog", "post-by-slug", params={"slug": slug}) if not post: return jsonify({"error": "not found"}), 404 resp = build_oembed_response( title=post["title"], oembed_type="link", thumbnail_url=post.get("feature_image"), url=events_url(f"/{post['slug']}"), ) return jsonify(resp) return app app = create_app()