Domain isolation: replace cross-domain imports with service calls
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 50s
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 50s
Replace direct Calendar, MarketPlace, and Post model queries with typed service calls (services.blog, services.calendar, services.market, services.cart). Blog registers all 4 services via domain_services_fn with has() guards for composable deployment. Key changes: - app.py: use domain_services_fn instead of inline service registration - admin routes: MarketPlace queries → services.market.marketplaces_for_container() - entry_associations: CalendarEntryPost → services.calendar.entry_ids_for_content() - markets service: Post query → services.blog.get_post_by_id/slug() - posts_data, post routes: use calendar/market/cart services - menu_items: glue imports → shared imports Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -11,9 +11,7 @@ from quart import (
|
||||
)
|
||||
from .services.post_data import post_data
|
||||
from .services.post_operations import toggle_post_like
|
||||
from shared.models.calendars import Calendar
|
||||
from shared.models.market_place import MarketPlace
|
||||
from sqlalchemy import select
|
||||
from shared.services.registry import services
|
||||
|
||||
from shared.browser.app.redis_cacher import cache_page, clear_cache
|
||||
|
||||
@@ -65,24 +63,11 @@ def register():
|
||||
p_data = getattr(g, "post_data", None)
|
||||
if p_data:
|
||||
from .services.entry_associations import get_associated_entries
|
||||
from shared.infrastructure.internal_api import get as api_get
|
||||
from shared.infrastructure.cart_identity import current_cart_identity
|
||||
|
||||
db_post_id = (g.post_data.get("post") or {}).get("id") # <-- integer
|
||||
calendars = (
|
||||
await g.s.execute(
|
||||
select(Calendar)
|
||||
.where(Calendar.container_type == "page", Calendar.container_id == db_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 == db_post_id, MarketPlace.deleted_at.is_(None))
|
||||
.order_by(MarketPlace.name.asc())
|
||||
)
|
||||
).scalars().all()
|
||||
calendars = await services.calendar.calendars_for_container(g.s, "page", db_post_id)
|
||||
markets = await services.market.marketplaces_for_container(g.s, "page", db_post_id)
|
||||
|
||||
# Fetch associated entries for nav display
|
||||
associated_entries = await get_associated_entries(g.s, db_post_id)
|
||||
@@ -95,20 +80,16 @@ def register():
|
||||
"associated_entries": associated_entries,
|
||||
}
|
||||
|
||||
# Page cart badge: fetch page-scoped cart count for pages
|
||||
# Page cart badge via service (replaces cross-app HTTP API)
|
||||
post_dict = p_data.get("post") or {}
|
||||
if post_dict.get("is_page"):
|
||||
page_cart = await api_get(
|
||||
"cart",
|
||||
f"/internal/cart/summary?page_slug={post_dict['slug']}",
|
||||
forward_session=True,
|
||||
ident = current_cart_identity()
|
||||
page_summary = await services.cart.cart_summary(
|
||||
g.s, user_id=ident["user_id"], session_id=ident["session_id"],
|
||||
page_slug=post_dict["slug"],
|
||||
)
|
||||
if page_cart:
|
||||
ctx["page_cart_count"] = page_cart.get("count", 0) + page_cart.get("calendar_count", 0)
|
||||
ctx["page_cart_total"] = page_cart.get("total", 0) + page_cart.get("calendar_total", 0)
|
||||
else:
|
||||
ctx["page_cart_count"] = 0
|
||||
ctx["page_cart_total"] = 0
|
||||
ctx["page_cart_count"] = page_summary.count + page_summary.calendar_count
|
||||
ctx["page_cart_total"] = float(page_summary.total + page_summary.calendar_total)
|
||||
|
||||
return ctx
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user