Domain isolation: replace cross-domain imports with service calls
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 55s

Replace direct Post, MarketPlace, Calendar model queries and HTTP API
calls with typed service calls. Events registers all 4 services via
domain_services_fn with has() guards.

Key changes:
- app.py: use domain_services_fn, Post/Calendar/MarketPlace queries
  → services.blog/calendar/market, HTTP cart API → services.cart
- calendars/markets services: Post → services.blog
- post_associations: Post → services.blog, direct queries → services
- markets routes: remove unused MarketPlace import
- glue imports → shared imports throughout

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
giles
2026-02-19 04:30:20 +00:00
parent 7b15f37686
commit f1b5aeac53
7 changed files with 82 additions and 89 deletions

55
app.py
View File

@@ -5,7 +5,6 @@ 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
@@ -17,34 +16,36 @@ 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
- cart_count/cart_total: via cart service (shared DB)
"""
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
from shared.services.navigation import get_navigation_tree
from shared.services.registry import services
from shared.infrastructure.cart_identity import current_cart_identity
ctx = await base_context()
ctx["menu_items"] = await get_navigation_tree(g.s)
# Cart data from cart API (includes both product + calendar counts)
cart_data = await api_get("cart", "/internal/cart/summary", forward_session=True)
if cart_data:
ctx["cart_count"] = cart_data.get("count", 0) + cart_data.get("calendar_count", 0)
ctx["cart_total"] = cart_data.get("total", 0) + cart_data.get("calendar_total", 0)
else:
ctx["cart_count"] = 0
ctx["cart_total"] = 0
# Cart data via service (replaces cross-app HTTP API)
ident = current_cart_identity()
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)
return ctx
def create_app() -> "Quart":
from shared.models.ghost_content import Post
from models.calendars import Calendar
from shared.models.market_place import MarketPlace
from services import register_domain_services
app = create_base_app("events", context_fn=events_context)
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")
@@ -90,11 +91,7 @@ def create_app() -> "Quart":
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()
post = await services.blog.get_post_by_slug(g.s, slug)
if not post:
abort(404)
g.post_data = {
@@ -114,20 +111,8 @@ def create_app() -> "Quart":
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()
calendars = await services.calendar.calendars_for_container(g.s, "page", post_id)
markets = await services.market.marketplaces_for_container(g.s, "page", post_id)
return {
**post_data,
"calendars": calendars,