b3d853ad35
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 4m48s
- fetch_fragment_batch() for N+1 avoidance with per-key Redis cache - link-card fragment handlers in blog, market, events, federation (single + batch mode) - link_card.html templates per app with content-specific previews - shared/infrastructure/oembed.py: build_oembed_response, build_og_meta, build_oembed_link_tag - GET /oembed routes on blog, market, events - og_meta + oembed_link rendering in base template <head> - INTERNAL_URL_ARTDAG in docker-compose.yml for cross-stack fragment fetches Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
240 lines
7.9 KiB
Python
240 lines
7.9 KiB
Python
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 sqlalchemy import select
|
|
|
|
from shared.infrastructure.factory import create_base_app
|
|
from shared.config import config
|
|
|
|
from bp import register_market_bp, register_all_markets, register_page_markets, register_fragments
|
|
|
|
|
|
async def market_context() -> dict:
|
|
"""
|
|
Market app context processor.
|
|
|
|
- nav_tree_html: fetched from blog as fragment
|
|
- cart_count/cart_total: via cart service (includes calendar entries)
|
|
- cart: direct ORM query (templates need .product relationship)
|
|
"""
|
|
from shared.infrastructure.context import base_context
|
|
from shared.services.navigation import get_navigation_tree
|
|
from shared.services.registry import services
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
from shared.infrastructure.fragments import fetch_fragments
|
|
from shared.models.market import CartItem
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
ctx = await base_context()
|
|
|
|
# Fallback for _nav.html when nav-tree fragment fetch fails
|
|
ctx["menu_items"] = await get_navigation_tree(g.s)
|
|
|
|
ident = current_cart_identity()
|
|
|
|
# cart_count/cart_total via service (consistent with blog/events apps)
|
|
summary = await services.cart.cart_summary(
|
|
g.s, user_id=ident["user_id"], session_id=ident["session_id"],
|
|
)
|
|
ctx["cart_count"] = summary.count + summary.calendar_count
|
|
ctx["cart_total"] = float(summary.total + summary.calendar_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": "market", "path": request.path}),
|
|
])
|
|
ctx["cart_mini_html"] = cart_mini_html
|
|
ctx["auth_menu_html"] = auth_menu_html
|
|
ctx["nav_tree_html"] = nav_tree_html
|
|
|
|
# ORM cart items for product templates (need .product relationship)
|
|
filters = [CartItem.deleted_at.is_(None)]
|
|
if ident["user_id"] is not None:
|
|
filters.append(CartItem.user_id == ident["user_id"])
|
|
elif ident["session_id"] is not None:
|
|
filters.append(CartItem.session_id == ident["session_id"])
|
|
else:
|
|
ctx["cart"] = []
|
|
return ctx
|
|
|
|
result = await g.s.execute(
|
|
select(CartItem).where(*filters).options(selectinload(CartItem.product))
|
|
)
|
|
ctx["cart"] = list(result.scalars().all())
|
|
|
|
return ctx
|
|
|
|
|
|
def create_app() -> "Quart":
|
|
from models.market_place import MarketPlace
|
|
from shared.services.registry import services
|
|
from services import register_domain_services
|
|
|
|
app = create_base_app(
|
|
"market",
|
|
context_fn=market_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 markets: / — global view across all pages
|
|
app.register_blueprint(
|
|
register_all_markets(),
|
|
url_prefix="/",
|
|
)
|
|
|
|
# Page markets: /<slug>/ — markets for a single page
|
|
app.register_blueprint(
|
|
register_page_markets(),
|
|
url_prefix="/<slug>",
|
|
)
|
|
|
|
# Market blueprint nested under post slug: /<page_slug>/<market_slug>/
|
|
app.register_blueprint(
|
|
register_market_bp(
|
|
url_prefix="/",
|
|
title=config()["market_title"],
|
|
),
|
|
url_prefix="/<page_slug>/<market_slug>",
|
|
)
|
|
|
|
app.register_blueprint(register_fragments())
|
|
|
|
# --- Auto-inject slugs into url_for() calls ---
|
|
@app.url_value_preprocessor
|
|
def pull_slugs(endpoint, values):
|
|
if values:
|
|
# page_markets blueprint uses "slug"
|
|
if "slug" in values:
|
|
g.post_slug = values.pop("slug")
|
|
# market blueprint uses "page_slug" / "market_slug"
|
|
if "page_slug" in values:
|
|
g.post_slug = values.pop("page_slug")
|
|
if "market_slug" in values:
|
|
g.market_slug = values.pop("market_slug")
|
|
|
|
@app.url_defaults
|
|
def inject_slugs(endpoint, values):
|
|
slug = g.get("post_slug")
|
|
if slug:
|
|
for param in ("slug", "page_slug"):
|
|
if param not in values and app.url_map.is_endpoint_expecting(endpoint, param):
|
|
values[param] = slug
|
|
market_slug = g.get("market_slug")
|
|
if market_slug and "market_slug" not in values:
|
|
if app.url_map.is_endpoint_expecting(endpoint, "market_slug"):
|
|
values["market_slug"] = market_slug
|
|
|
|
# --- Load post and market data ---
|
|
@app.before_request
|
|
async def hydrate_market():
|
|
post_slug = getattr(g, "post_slug", None)
|
|
market_slug = getattr(g, "market_slug", None)
|
|
if not post_slug:
|
|
return
|
|
|
|
# Load post by slug via blog service
|
|
post = await services.blog.get_post_by_slug(g.s, post_slug)
|
|
if not post:
|
|
abort(404)
|
|
|
|
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,
|
|
},
|
|
}
|
|
|
|
# Only load market when market_slug is present (/<page_slug>/<market_slug>/)
|
|
if not market_slug:
|
|
return
|
|
|
|
market = (
|
|
await g.s.execute(
|
|
select(MarketPlace).where(
|
|
MarketPlace.slug == market_slug,
|
|
MarketPlace.container_type == "page",
|
|
MarketPlace.container_id == post.id,
|
|
MarketPlace.deleted_at.is_(None),
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if not market:
|
|
abort(404)
|
|
g.market = market
|
|
|
|
@app.context_processor
|
|
async def inject_post():
|
|
post_data = getattr(g, "post_data", None)
|
|
if not post_data:
|
|
return {}
|
|
return {**post_data}
|
|
|
|
# --- oEmbed endpoint ---
|
|
@app.get("/oembed")
|
|
async def oembed():
|
|
from urllib.parse import urlparse
|
|
from quart import jsonify
|
|
from shared.models.market import Product
|
|
from shared.infrastructure.urls import market_url
|
|
from shared.infrastructure.oembed import build_oembed_response
|
|
|
|
url = request.args.get("url", "")
|
|
if not url:
|
|
return jsonify({"error": "url parameter required"}), 400
|
|
|
|
parsed = urlparse(url)
|
|
# Market product URLs: /.../<page_slug>/<market_slug>/product/<slug>/
|
|
parts = [p for p in parsed.path.strip("/").split("/") if p]
|
|
slug = ""
|
|
for i, part in enumerate(parts):
|
|
if part == "product" and i + 1 < len(parts):
|
|
slug = parts[i + 1]
|
|
break
|
|
if not slug:
|
|
return jsonify({"error": "could not extract product slug"}), 404
|
|
|
|
product = (
|
|
await g.s.execute(select(Product).where(Product.slug == slug))
|
|
).scalar_one_or_none()
|
|
if not product:
|
|
return jsonify({"error": "not found"}), 404
|
|
|
|
resp = build_oembed_response(
|
|
title=product.title or slug,
|
|
oembed_type="link",
|
|
thumbnail_url=product.image,
|
|
url=market_url(f"/product/{product.slug}/"),
|
|
)
|
|
return jsonify(resp)
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|