from __future__ import annotations import path_setup # noqa: F401 # adds shared_lib to sys.path from pathlib import Path from quart import g, request from jinja2 import FileSystemLoader, ChoiceLoader from sqlalchemy import select from shared.infrastructure.factory import create_base_app from shared.config import config from shared.models import KV from bp import ( register_auth_bp, register_blog_bp, register_admin, register_menu_items, register_snippets, register_coop_api, ) async def coop_context() -> dict: """ Coop app context processor. - menu_items: direct DB query (coop owns this data) - cart_count/cart_total: fetched from cart internal API """ from shared.infrastructure.context import base_context from bp.menu_items.services.menu_items import get_all_menu_items from shared.infrastructure.internal_api import get as api_get ctx = await base_context() # Coop owns menu_items — query directly ctx["menu_items"] = await get_all_menu_items(g.s) # Cart data from cart app 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": app = create_base_app("coop", context_fn=coop_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, ]) # --- blueprints --- app.register_blueprint(register_auth_bp()) app.register_blueprint( register_blog_bp( url_prefix=config()["blog_root"], title=config()["blog_title"], ), url_prefix=config()["blog_root"], ) app.register_blueprint(register_admin("/settings")) app.register_blueprint(register_menu_items()) app.register_blueprint(register_snippets()) # Internal API (server-to-server, CSRF-exempt) app.register_blueprint(register_coop_api()) # --- KV admin endpoints --- @app.get("/settings/kv/") async def kv_get(key: str): row = ( await g.s.execute(select(KV).where(KV.key == key)) ).scalar_one_or_none() return {"key": key, "value": (row.value if row else None)} @app.post("/settings/kv/") async def kv_set(key: str): data = await request.get_json() or {} val = data.get("value", "") obj = await g.s.get(KV, key) if obj is None: obj = KV(key=key, value=val) g.s.add(obj) else: obj.value = val return {"ok": True, "key": key, "value": val} # --- debug: url rules --- @app.get("/__rules") async def dump_rules(): rules = [] for r in app.url_map.iter_rules(): rules.append({ "endpoint": r.endpoint, "rule": repr(r.rule), "methods": sorted(r.methods - {"HEAD", "OPTIONS"}), "strict_slashes": r.strict_slashes, }) return {"rules": rules} return app app = create_app()