from __future__ import annotations import path_setup # noqa: F401 # adds shared_lib to sys.path from pathlib import Path from quart import g, abort, render_template, make_response from jinja2 import FileSystemLoader, ChoiceLoader from sqlalchemy import select from shared.factory import create_base_app from config import config from suma_browser.app.bp import register_market_bp async def market_context() -> dict: """ Market app context processor. - menu_items: fetched from coop internal API - cart_count/cart_total: fetched from cart internal API """ from shared.context import base_context from shared.internal_api import get as api_get, dictobj ctx = await base_context() # Menu items from coop API (wrapped for attribute access in templates) menu_data = await api_get("coop", "/internal/menu-items") ctx["menu_items"] = dictobj(menu_data) if menu_data else [] # 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 models.market_place import MarketPlace from models.ghost_content import Post app = create_base_app("market", context_fn=market_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, ]) # Market blueprint scoped under // app.register_blueprint( register_market_bp( url_prefix="/", title=config()["coop_title"], ), url_prefix="/", ) # --- Auto-inject market_slug into url_for() calls --- @app.url_value_preprocessor def pull_market_slug(endpoint, values): if values and "market_slug" in values: g.market_slug = values.pop("market_slug") @app.url_defaults def inject_market_slug(endpoint, values): slug = g.get("market_slug") if slug and "market_slug" not in values: if app.url_map.is_endpoint_expecting(endpoint, "market_slug"): values["market_slug"] = slug # --- Load market data for market_slug --- @app.before_request async def hydrate_market(): slug = getattr(g, "market_slug", None) if not slug: return market = ( await g.s.execute( select(MarketPlace).where( MarketPlace.slug == slug, MarketPlace.deleted_at.is_(None), ) ) ).scalar_one_or_none() if not market: abort(404) g.market = market # Load associated Post for context post = ( await g.s.execute( select(Post).where(Post.id == market.post_id) ) ).scalar_one_or_none() if post: g.post_data = { "post": { "id": post.id, "title": post.title, "slug": post.slug, "feature_image": post.feature_image, "html": post.html, "status": post.status, "visibility": post.visibility, "is_page": post.is_page, }, } # --- Root route: market listing --- @app.get("/") async def markets_listing(): markets = ( await g.s.execute( select(MarketPlace) .where(MarketPlace.deleted_at.is_(None)) .order_by(MarketPlace.name) ) ).scalars().all() html = await render_template( "_types/market/markets_listing.html", markets=markets, ) return await make_response(html) return app app = create_app()