Decouple blog models and BlogService from shared layer

Move Post/Author/Tag/PostAuthor/PostTag/PostUser models from
shared/models/ghost_content.py to blog/models/content.py so blog-domain
models no longer live in the shared layer. Replace the shared
SqlBlogService + BlogService protocol with a blog-local singleton
(blog_service), and switch entry_associations.py from direct DB access
to HTTP fetch_data("blog", "post-by-id") to respect the inter-service
boundary.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

shared/services/blog_impl.py

@@ -1,65 +0,0 @@
-"""SQL-backed BlogService implementation.
-
-Queries ``shared.models.ghost_content.Post`` — only this module may read
-blog-domain tables on behalf of other domains.
-"""
-from __future__ import annotations
-
-from sqlalchemy import select, func
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from shared.models.ghost_content import Post
-from shared.contracts.dtos import PostDTO
-
-
-def _post_to_dto(post: Post) -> PostDTO:
-    return PostDTO(
-        id=post.id,
-        slug=post.slug,
-        title=post.title,
-        status=post.status,
-        visibility=post.visibility,
-        is_page=post.is_page,
-        feature_image=post.feature_image,
-        html=post.html,
-        excerpt=post.excerpt,
-        custom_excerpt=post.custom_excerpt,
-        published_at=post.published_at,
-    )
-
-
-class SqlBlogService:
-    async def get_post_by_slug(self, session: AsyncSession, slug: str) -> PostDTO | None:
-        post = (
-            await session.execute(select(Post).where(Post.slug == slug))
-        ).scalar_one_or_none()
-        return _post_to_dto(post) if post else None
-
-    async def get_post_by_id(self, session: AsyncSession, id: int) -> PostDTO | None:
-        post = (
-            await session.execute(select(Post).where(Post.id == id))
-        ).scalar_one_or_none()
-        return _post_to_dto(post) if post else None
-
-    async def get_posts_by_ids(self, session: AsyncSession, ids: list[int]) -> list[PostDTO]:
-        if not ids:
-            return []
-        result = await session.execute(select(Post).where(Post.id.in_(ids)))
-        return [_post_to_dto(p) for p in result.scalars().all()]
-
-    async def search_posts(
-        self, session: AsyncSession, query: str, page: int = 1, per_page: int = 10,
-    ) -> tuple[list[PostDTO], int]:
-        """Search posts by title with pagination. Not part of the Protocol
-        (admin-only use in events), but provided for convenience."""
-        if query:
-            count_stmt = select(func.count(Post.id)).where(Post.title.ilike(f"%{query}%"))
-            posts_stmt = select(Post).where(Post.title.ilike(f"%{query}%")).order_by(Post.title)
-        else:
-            count_stmt = select(func.count(Post.id))
-            posts_stmt = select(Post).order_by(Post.published_at.desc().nullslast())
-
-        total = (await session.execute(count_stmt)).scalar() or 0
-        offset = (page - 1) * per_page
-        result = await session.execute(posts_stmt.limit(per_page).offset(offset))
-        return [_post_to_dto(p) for p in result.scalars().all()], total

shared/services/entry_associations.py

@@ -4,16 +4,12 @@ Only uses HTTP-based fetch_data/call_action, no direct DB access.
 """
 from __future__ import annotations
 
-from sqlalchemy.ext.asyncio import AsyncSession
-
 from shared.infrastructure.actions import call_action, ActionError
 from shared.infrastructure.data_client import fetch_data
 from shared.contracts.dtos import CalendarEntryDTO, dto_from_dict
-from shared.services.registry import services
 
 
 async def toggle_entry_association(
-    session: AsyncSession,
     post_id: int,
     entry_id: int
 ) -> tuple[bool, str | None]:
@@ -21,7 +17,7 @@ async def toggle_entry_association(
     Toggle association between a post and calendar entry.
     Returns (is_now_associated, error_message).
     """
-    post = await services.blog.get_post_by_id(session, post_id)
+    post = await fetch_data("blog", "post-by-id", params={"id": post_id}, required=False)
     if not post:
         return False, "Post not found"
 
@@ -35,7 +31,6 @@ async def toggle_entry_association(
 
 
 async def get_post_entry_ids(
-    session: AsyncSession,
     post_id: int
 ) -> set[int]:
     """
@@ -49,7 +44,6 @@ async def get_post_entry_ids(
 
 
 async def get_associated_entries(
-    session: AsyncSession,
     post_id: int,
     page: int = 1,
     per_page: int = 10

shared/services/registry.py

@@ -8,15 +8,14 @@ Usage::
     from shared.services.registry import services
 
     # Register at app startup (own domain only)
-    services.blog = SqlBlogService()
+    services.calendar = SqlCalendarService()
 
     # Use locally within the owning app
-    post = await services.blog.get_post_by_slug(session, slug)
+    cals = await services.calendar.calendars_for_container(session, "page", page_id)
 """
 from __future__ import annotations
 
 from shared.contracts.protocols import (
-    BlogService,
     CalendarService,
     MarketService,
     CartService,
@@ -33,23 +32,11 @@ class _ServiceRegistry:
     """
 
     def __init__(self) -> None:
-        self._blog: BlogService | None = None
         self._calendar: CalendarService | None = None
         self._market: MarketService | None = None
         self._cart: CartService | None = None
         self._federation: FederationService | None = None
 
-    # -- blog -----------------------------------------------------------------
-    @property
-    def blog(self) -> BlogService:
-        if self._blog is None:
-            raise RuntimeError("BlogService not registered")
-        return self._blog
-
-    @blog.setter
-    def blog(self, impl: BlogService) -> None:
-        self._blog = impl
-
     # -- calendar -------------------------------------------------------------
     @property
     def calendar(self) -> CalendarService: