Public store

Autodoc for the dual public store and its shared shapers. See Store design for the narrative.

Store dependency

store

Public store dependency wiring.

get_public_store()

Return the production SQL-backed public store (override in tests).

Source code in src/cms_api/public/store/__init__.py

def get_public_store() -> PublicStore:
    """Return the production SQL-backed public store (override in tests)."""
    from .sql import SqlPublicStore

    return SqlPublicStore()

SQL store

sql

SQLAlchemy-backed public read store for production.

SqlPublicStore

Production public persistence using hq.sql_models ORM entities.

Source code in src/cms_api/public/store/sql.py

class SqlPublicStore:
    """Production public persistence using hq.sql_models ORM entities."""

    def resolve_site(self, site_acronym: str) -> SiteOut:
        """Return an active site by acronym or domain."""
        from hq.sql_models.core import Site

        with session_scope() as session:
            row = (
                session.query(Site)
                .filter((Site.acronym == site_acronym) | (Site.domain == site_acronym))
                .one_or_none()
            )
        if not row or not row.is_active:
            raise APIError("resource_not_found", "Site not found", 404, {"site_acronym": site_acronym})
        return SiteOut(
            id=row.id,
            acronym=row.acronym,
            name=row.name,
            tagline=row.tagline,
            description=row.description,
            domain=row.domain,
            default_language=row.default_language,
            meta_title=row.meta_title,
            meta_description=row.meta_description,
            template_id=row.template_id,
            is_active=bool(row.is_active),
        )

    def _resolve_site_row(self, site_acronym: str):
        """Return the ORM site row (internal)."""
        from hq.sql_models.core import Site

        with session_scope() as session:
            row = (
                session.query(Site)
                .filter((Site.acronym == site_acronym) | (Site.domain == site_acronym))
                .one_or_none()
            )
        if not row or not row.is_active:
            raise APIError("resource_not_found", "Site not found", 404, {"site_acronym": site_acronym})
        return row

    def resolve_site_any(self, site: str) -> SiteOut:
        """Resolve a site by numeric id first, then acronym/domain. Used by v2."""
        from hq.sql_models.core import Site

        with session_scope() as session:
            row = None
            try:
                site_id = int(site)
            except (TypeError, ValueError):
                site_id = None
            if site_id is not None:
                row = session.get(Site, site_id)
            if row is None:
                row = (
                    session.query(Site)
                    .filter((Site.acronym == site) | (Site.domain == site))
                    .one_or_none()
                )
        if not row or not row.is_active:
            raise APIError("resource_not_found", "Site not found", 404, {"site": site})
        return SiteOut(
            id=row.id,
            acronym=row.acronym,
            name=row.name,
            tagline=row.tagline,
            description=row.description,
            domain=row.domain,
            default_language=row.default_language,
            meta_title=row.meta_title,
            meta_description=row.meta_description,
            template_id=row.template_id,
            is_active=bool(row.is_active),
        )

    def _effective_settings_v2(self, session, site_id: int) -> Dict[str, Any]:
        """Merge per-site and global (-1) EAV settings."""
        merged = effective_settings(session, site_id)
        merged["_settings_updated_at"] = settings_updated_at(session, site_id)
        return merged

    def _load_settings_row(self, session, site_id: int):
        """Return None; wide rows are replaced by EAV (kept for script compat)."""
        return None

    def _effective_settings(self, session, site_id: int) -> Dict[str, Any]:
        """Merge per-site settings with global defaults (site_id = -1)."""
        return effective_settings(session, site_id)

    def _site_summary(self, site) -> PublicSiteSummary:
        return PublicSiteSummary(
            id=site.id,
            acronym=site.acronym,
            name=site.name,
            domain=site.domain,
            default_language=site.default_language,
            tagline=site.tagline,
        )

    def get_public_config(self, site_acronym: str) -> Tuple[PublicSiteConfig, Optional[datetime]]:
        """Return public site configuration."""
        site = self.resolve_site(site_acronym)
        with session_scope() as session:
            settings = self._effective_settings(session, site.id)
        config = PublicSiteConfig(
            site=self._site_summary(site),
            default_meta_title=settings.get("default_meta_title"),
            default_meta_description=settings.get("default_meta_description"),
            default_og_image=absolutize_media_url(settings.get("default_og_image"), settings.get("cdn_base_url")),
            default_favicon=absolutize_media_url(settings.get("default_favicon"), settings.get("cdn_base_url")),
            cdn_base_url=settings.get("cdn_base_url"),
            maintenance_mode=bool(settings.get("maintenance_mode")),
            footer_text=settings.get("footer_text"),
            navigation=settings.get("navigation"),
            icon_url=absolutize_media_url(settings.get("icon_url"), settings.get("cdn_base_url")),
        )
        return config, datetime.now(timezone.utc)

    def _fetch_published_content(self, session, site_id: int, slug: str, language: Optional[str]):
        from hq.sql_models.content import Content

        query = session.query(Content).filter(
            Content.site_id == site_id,
            Content.slug == slug,
            Content.status == _PUBLISHED,
            Content.is_active.is_(True),
        )
        if language:
            query = query.filter(Content.language == language)
        row = query.one_or_none()
        if not row:
            raise APIError("resource_not_found", "Content not found", 404, {"slug": slug})
        return row

    def _content_public(self, row, cdn_base_url: Optional[str]) -> PublicContentOut:
        return PublicContentOut(
            id=row.id,
            site_id=row.site_id,
            content_type=row.content_type,
            language=row.language,
            slug=row.slug,
            title=row.title,
            description=row.description,
            author_id=row.author_id,
            featured_image=absolutize_media_url(row.featured_image, cdn_base_url),
            show_in_homepage=bool(row.show_in_homepage),
            is_featured=bool(row.is_featured),
            is_trending=bool(row.is_trending),
            content_blocks=row.content_blocks,
            settings=row.settings,
            published_at=row.published_at,
            categories=[],
            tags=[],
        )

    def _author_public(self, session, author_id: int, cdn_base_url: Optional[str]) -> PublicAuthorOut:
        from hq.sql_models.core import Author

        row = session.get(Author, author_id)
        if not row or not row.is_active:
            raise APIError("resource_not_found", "Author not found", 404, {"author_id": author_id})
        return PublicAuthorOut(
            id=row.id,
            slug=row.slug,
            name=row.name,
            initials=row.initials,
            image=absolutize_media_url(row.image, cdn_base_url),
            bio=row.bio,
        )

    def _seo_out(self, row, cdn_base_url: Optional[str]) -> SeoOut:
        return SeoOut(
            slug=row.slug,
            seo_title=row.seo_title or row.title,
            seo_description=row.seo_description or row.description,
            og_title=row.og_title,
            og_description=row.og_description,
            og_image=absolutize_media_url(row.og_image or row.featured_image, cdn_base_url),
            canonical_url=row.canonical_url,
            noindex=bool(row.noindex),
            schema_type=row.schema_type,
        )

    def _resolved_blocks(self, session, site) -> List[TemplateBlockPublic]:
        from hq.sql_models.core import SiteBlockOverride
        from hq.sql_models.template import TemplateBlock

        if not site.template_id:
            return []
        overrides = {
            row.element_type: row
            for row in session.query(SiteBlockOverride).filter(SiteBlockOverride.site_id == site.id).all()
        }
        blocks: List[TemplateBlockPublic] = []
        for row in session.query(TemplateBlock).filter(TemplateBlock.template_id == site.template_id).all():
            override = overrides.get(row.element_type)
            blocks.append(
                TemplateBlockPublic(
                    element_type=row.element_type,
                    html_content=override.html_content if override else row.html_content,
                    settings=override.settings if override and override.settings is not None else row.settings,
                )
            )
        return blocks

    def get_published_content(
        self, site_acronym: str, slug: str, language: Optional[str]
    ) -> Tuple[PublicContentOut, Optional[datetime]]:
        """Return a published content document."""
        site = self.resolve_site(site_acronym)
        with session_scope() as session:
            row = self._fetch_published_content(session, site.id, slug, language)
            settings = self._effective_settings(session, site.id)
            return self._content_public(row, settings.get("cdn_base_url")), row.published_at

    def get_seo(self, site_acronym: str, slug: str, language: Optional[str]) -> Tuple[SeoOut, Optional[datetime]]:
        """Return SEO metadata for a published slug."""
        site = self.resolve_site(site_acronym)
        with session_scope() as session:
            row = self._fetch_published_content(session, site.id, slug, language)
            settings = self._effective_settings(session, site.id)
            return self._seo_out(row, settings.get("cdn_base_url")), row.published_at

    def build_page_payload(
        self, site_acronym: str, slug: str, language: Optional[str]
    ) -> Tuple[PageRenderOut, Optional[datetime]]:
        """Build the full page render payload."""
        site = self.resolve_site(site_acronym)
        with session_scope() as session:
            row = self._fetch_published_content(session, site.id, slug, language)
            settings = self._effective_settings(session, site.id)
            cdn = settings.get("cdn_base_url")
            site_summary = self._site_summary(site)
            content = self._content_public(row, cdn)
            author = self._author_public(session, row.author_id, cdn)
            seo = self._seo_out(row, cdn)
            blocks = self._resolved_blocks(session, site)
            data = {
                "site": site_summary.model_dump(),
                "content": content.model_dump(),
                "author": author.model_dump(),
            }
            payload = PageRenderOut(
                site=site_summary,
                content=content,
                author=author,
                template={"blocks": blocks},
                seo=seo,
                data=data,
                meta=PageMeta(bindings=collect_bindings(data)),
            )
            return payload, row.published_at

    def list_published_content(
        self,
        site_acronym: str,
        content_type: Optional[str],
        featured: Optional[bool],
        homepage: Optional[bool],
        language: Optional[str],
        limit: int,
        cursor: Optional[str],
    ) -> Tuple[List[PublicContentOut], Optional[str], Optional[datetime]]:
        """Return a cursor-paginated list of published content."""
        from hq.sql_models.content import Content

        site = self.resolve_site(site_acronym)
        offset = cursor_offset(cursor)
        with session_scope() as session:
            query = session.query(Content).filter(
                Content.site_id == site.id,
                Content.status == _PUBLISHED,
                Content.is_active.is_(True),
            )
            if content_type:
                query = query.filter(Content.content_type == content_type)
            if featured is not None:
                query = query.filter(Content.is_featured == featured)
            if homepage is not None:
                query = query.filter(Content.show_in_homepage == homepage)
            if language:
                query = query.filter(Content.language == language)
            rows = query.order_by(Content.id).offset(offset).limit(limit + 1).all()
            settings = self._effective_settings(session, site.id)
            cdn = settings.get("cdn_base_url")
        next_cursor = offset_page_cursor(offset, limit, len(rows))
        page = rows[:limit]
        items = [self._content_public(row, cdn) for row in page]
        last_mod = max((row.published_at for row in page if row.published_at), default=None)
        return items, next_cursor, last_mod

    def list_site_categories(self, site_acronym: str) -> Tuple[List[SiteCategoryPublic], Optional[datetime]]:
        """List enabled categories for a site."""
        from hq.sql_models.taxonomy import SiteCategory

        site = self.resolve_site(site_acronym)
        with session_scope() as session:
            rows = (
                session.query(SiteCategory)
                .filter(SiteCategory.site_id == site.id)
                .order_by(SiteCategory.sort_order, SiteCategory.category)
                .all()
            )
        return [
            SiteCategoryPublic(id=row.id, category=row.category, sort_order=row.sort_order or 0) for row in rows
        ], datetime.now(timezone.utc)

    # --- v2 worker-facing surface (AISD-116) -------------------------------

    def _site_categories_light(self, session, site_id: int) -> List[Tuple[int, str, str, int]]:
        from hq.sql_models.lookup import Category
        from hq.sql_models.taxonomy import SiteCategory

        rows = (
            session.query(SiteCategory, Category)
            .outerjoin(Category, Category.name == SiteCategory.category)
            .filter(SiteCategory.site_id == site_id, SiteCategory.is_active.is_(True))
            .order_by(SiteCategory.sort_order, SiteCategory.category)
            .all()
        )
        result: List[Tuple[int, str, str, int]] = []
        for site_cat, category in rows:
            result.append(
                (
                    category.id if category else site_cat.id,
                    site_cat.category,
                    category.name if category else site_cat.category,
                    site_cat.sort_order or 0,
                )
            )
        return result

    def _site_tags_light(self, session, site_id: int) -> List[Tuple[int, str, str]]:
        from hq.sql_models.content import Content
        from hq.sql_models.lookup import Tag
        from hq.sql_models.taxonomy import ContentTag

        rows = (
            session.query(Tag.id, Tag.name)
            .join(ContentTag, ContentTag.tag == Tag.name)
            .join(Content, Content.id == ContentTag.content_id)
            .filter(Content.site_id == site_id, Content.status == _PUBLISHED)
            .distinct()
            .all()
        )
        return [(tag_id, name, name) for (tag_id, name) in rows]

    def _site_authors_light(self, session, site_id: int) -> List[Dict[str, Any]]:
        """All active authors for a site + their published-article counts (for /authors)."""
        from sqlalchemy import func

        from hq.sql_models.content import Content
        from hq.sql_models.core import Author

        counts = dict(
            session.query(Content.author_id, func.count(Content.id))
            .filter(
                Content.site_id == site_id,
                Content.status == _PUBLISHED,
                Content.is_active.is_(True),
            )
            .group_by(Content.author_id)
            .all()
        )
        rows = (
            session.query(Author)
            .filter(Author.site_id == site_id, Author.is_active.is_(True))
            .order_by(Author.name)
            .all()
        )
        return [
            {
                "id": a.id,
                "slug": a.slug,
                "name": a.name,
                "initials": a.initials,
                "image": a.image,
                "bio": a.bio,
                "article_count": int(counts.get(a.id, 0)),
            }
            for a in rows
        ]

    def _resolved_templates_map(
        self, session, site
    ) -> Dict[str, List[Tuple[Optional[str], Optional[Any]]]]:
        from hq.sql_models.lookup import TemplateElement
        from hq.sql_models.template import SiteBlockOverride, TemplateBlock

        elements = {
            element.name: element
            for element in session.query(TemplateElement).filter(TemplateElement.is_active.is_(True)).all()
        }
        blocks = {
            row.element_type: row
            for row in session.query(TemplateBlock)
            .filter(TemplateBlock.template_id == site.template_id, TemplateBlock.is_active.is_(True))
            .all()
        } if site.template_id else {}
        overrides = {
            row.element_type: row
            for row in session.query(SiteBlockOverride)
            .filter(SiteBlockOverride.site_id == site.id, SiteBlockOverride.is_active.is_(True))
            .all()
        }
        resolved: Dict[str, List[Tuple[Optional[str], Optional[Any]]]] = {}
        # Union of every element_type present across element defaults, template
        # blocks, and site overrides -- an override/block with no registered
        # element default must still surface (precedence: override > block > default).
        element_names = set(elements) | set(blocks) | set(overrides)
        for name in sorted(element_names):
            element = elements.get(name)
            html = element.html_content if element is not None else None
            settings = element.default_settings if element is not None else None
            block = blocks.get(name)
            override = overrides.get(name)
            if block is not None:
                if block.html_content is not None:
                    html = block.html_content
                if block.settings is not None:
                    settings = block.settings
            if override is not None:
                if override.html_content is not None:
                    html = override.html_content
                if override.settings is not None:
                    settings = override.settings
            resolved[name] = [(html, settings)]
        return resolved

    def _active_template(
        self, session, site
    ) -> Optional[Tuple[int, str, Optional[str], Optional[str]]]:
        from hq.sql_models.template import Template

        if not site.template_id:
            return None
        template = session.get(Template, site.template_id)
        if template is None:
            return None
        # getattr-defensive: the deployed hq model may predate the css/css_version
        # columns (the DB has them, but an older ORM model won't declare them).
        # Reading missing attrs must not 500 /settings.
        return (
            template.id,
            template.slug,
            template.version,
            getattr(template, "css_version", None),
        )

    def _resolved_page_types(self, session, site) -> List[Dict[str, Any]]:
        """Return the active template's page descriptors as raw dicts (P1.b).

        Ordered by ``position`` (NULLs last) then ``type``/``variant`` for a
        stable worker match order; validated/coerced by the shared shaper.
        """
        from hq.sql_models.template import PageType

        if not site.template_id:
            return []
        rows = (
            session.query(PageType)
            .filter(PageType.template_id == site.template_id, PageType.is_active.is_(True))
            .all()
        )
        descriptors: List[Dict[str, Any]] = []
        for row in rows:
            descriptors.append(
                {
                    "type": row.type,
                    "variant": row.variant,
                    "routes": row.routes,
                    "layout": row.layout,
                    "ttl": row.ttl,
                    "sources": row.sources or [],
                    "bindings": row.bindings or {},
                    "grids": row.grids or {},
                    "sidebar": row.sidebar,
                    "partial": row.partial,
                    "seo": row.seo,
                    "transforms": row.transforms or [],
                    "builder": row.builder,
                    "position": row.position,
                }
            )
        return descriptors

    def get_template_css(self, slug: str) -> Tuple[str, Optional[str]]:
        """Return ``(css, css_version)`` for a template by slug (for the CSS endpoint)."""
        from hq.sql_models.template import Template

        with session_scope() as session:
            row = (
                session.query(Template)
                .filter(Template.slug == slug, Template.is_active.is_(True))
                .one_or_none()
            )
            if row is None:
                raise APIError("resource_not_found", "Template not found", 404, {"slug": slug})
            # Read inside the session (avoid DetachedInstanceError) + getattr-defensive.
            return (getattr(row, "css", None) or "", getattr(row, "css_version", None))

    def build_settings_blob(self, site: str, language: Optional[str] = None) -> Tuple[SettingsBundle, Optional[datetime]]:
        """Build the v2 settings bundle (identity + merged settings + templates)."""
        resolved_site = self.resolve_site_any(site)
        with session_scope() as session:
            settings = self._effective_settings_v2(session, resolved_site.id)
            categories = self._site_categories_light(session, resolved_site.id)
            tags = self._site_tags_light(session, resolved_site.id)
            templates = self._resolved_templates_map(session, resolved_site)
            active_template = self._active_template(session, resolved_site)
            page_types = self._resolved_page_types(session, resolved_site)
            authors = self._site_authors_light(session, resolved_site.id)
        last_modified = settings.get("_settings_updated_at")
        bundle = v2_common.build_settings_blob(
            site=resolved_site,
            settings=settings,
            categories=categories,
            tags=tags,
            templates=templates,
            active_template=active_template,
            feature_flags={},
            last_modified=last_modified,
            page_types=page_types,
            authors=authors,
        )
        return bundle, last_modified

    def _content_by_slug(self, session, site_id: int, slug: str, language: Optional[str]):
        from hq.sql_models.content import Content

        query = session.query(Content).filter(
            Content.site_id == site_id,
            Content.slug == slug,
            Content.content_type.in_(v2_common.ARTICLE_CONTENT_TYPES),
            Content.status == _PUBLISHED,
            Content.is_active.is_(True),
        )
        if language:
            query = query.filter(Content.language == language)
        return query.one_or_none()

    def _content_taxonomy(self, session, content_id: int) -> Tuple[List[str], List[str]]:
        from hq.sql_models.taxonomy import ContentCategory, ContentTag

        categories = [
            row.category
            for row in session.query(ContentCategory).filter(ContentCategory.content_id == content_id).all()
        ]
        tags = [
            row.tag for row in session.query(ContentTag).filter(ContentTag.content_id == content_id).all()
        ]
        return categories, tags

    def _history_row(self, session, content_id: int, version: int, live_row):
        """Return a (merged snapshot, changed_at) tuple for ``version``.

        The merge keeps the live row's identity columns and overlays only the
        snapshot's mutable business columns; ``changed_at`` is the snapshot's own
        change timestamp (None when the column is unset).
        """
        from hq.sql_models.content import ContentHistory

        history_rows = (
            session.query(ContentHistory)
            .filter(ContentHistory.content_id == content_id)
            .order_by(ContentHistory.id)
            .all()
        )
        if version < 1 or version > len(history_rows):
            raise APIError("resource_not_found", "Content version not found", 404, {"version": version})
        snapshot = history_rows[version - 1]
        return _MergedContent(snapshot, live_row), getattr(snapshot, "changed_at", None)

    def _related_articles(self, session, site_id, content_id, categories, limit, cdn):
        from hq.sql_models.content import Content
        from hq.sql_models.taxonomy import ContentCategory

        if not categories:
            return []
        related_ids = (
            session.query(ContentCategory.content_id)
            .filter(ContentCategory.category.in_(categories), ContentCategory.content_id != content_id)
            .distinct()
            .all()
        )
        ids = [row[0] for row in related_ids]
        if not ids:
            return []
        rows = (
            session.query(Content)
            .filter(
                Content.id.in_(ids),
                Content.site_id == site_id,
                Content.content_type.in_(v2_common.ARTICLE_CONTENT_TYPES),
                Content.status == _PUBLISHED,
                Content.is_active.is_(True),
            )
            .order_by(Content.published_at.desc(), Content.id.desc())
            .limit(limit)
            .all()
        )
        author_names = self._author_names(session, [r.author_id for r in rows])
        return [
            v2_common.related_item(
                id=r.id,
                slug=r.slug,
                title=r.title,
                featured_image=r.featured_image,
                author_name=author_names.get(r.author_id),
                published_at=r.published_at,
                cdn_base_url=cdn,
            )
            for r in rows
        ]

    def _author_names(self, session, author_ids) -> Dict[int, str]:
        from hq.sql_models.core import Author

        unique_ids = list({aid for aid in author_ids if aid is not None})
        if not unique_ids:
            return {}
        rows = session.query(Author.id, Author.name).filter(Author.id.in_(unique_ids)).all()
        return {aid: name for (aid, name) in rows}

    def _authors_light(self, session, author_ids) -> Dict[int, Tuple[str, Optional[str]]]:
        """Return id -> (name, image) for the given authors in one batched query."""
        from hq.sql_models.core import Author

        unique_ids = list({aid for aid in author_ids if aid is not None})
        if not unique_ids:
            return {}
        rows = (
            session.query(Author.id, Author.name, Author.image)
            .filter(Author.id.in_(unique_ids))
            .all()
        )
        return {aid: (name, image) for (aid, name, image) in rows}

    def get_article_page(
        self,
        site_id: str,
        slug: str,
        language: Optional[str] = None,
        version: Optional[int] = None,
        related_limit: int = 6,
    ) -> Tuple[ArticlePage, Optional[datetime]]:
        """Build the v2 article page payload, optionally from content_history."""
        from hq.sql_models.core import Author

        site = self.resolve_site_any(site_id)
        with session_scope() as session:
            live_row = self._content_by_slug(session, site.id, slug, language)
            if live_row is None:
                raise APIError("resource_not_found", "Content not found", 404, {"slug": slug})
            snapshot_changed_at: Optional[datetime] = None
            if version:
                row, snapshot_changed_at = self._history_row(session, live_row.id, version, live_row)
            else:
                row = live_row
            settings = self._effective_settings_v2(session, site.id)
            cdn = settings.get("cdn_base_url")
            categories, tags = self._content_taxonomy(session, live_row.id)
            author = session.get(Author, row.author_id)
            if author is None:
                raise APIError("resource_not_found", "Author not found", 404, {"author_id": row.author_id})
            content = v2_common.build_article_content(row, categories, tags, cdn)
            author_slice = v2_common.author_slice(author, cdn)
            seo = v2_common.build_article_seo(row, author.name, cdn)
            breadcrumbs = v2_common.build_breadcrumbs(
                site.name, categories[0] if categories else None, content.title
            )
            related = self._related_articles(session, site.id, live_row.id, categories, related_limit, cdn)
        # On a versioned read the snapshot's own change timestamp is the resource's
        # last-modified; fall back to the live row only when changed_at is null.
        if version:
            last_modified = snapshot_changed_at or content.updated_at or content.published_at
        else:
            last_modified = content.updated_at or content.published_at
        page = ArticlePage(
            content=content,
            author=author_slice,
            seo=seo,
            breadcrumbs=breadcrumbs,
            related=related,
            meta=v2_common.meta_envelope("pages", last_modified),
        )
        return page, last_modified

    def get_author_profile(self, site_id: str, slug: str) -> Tuple[AuthorProfile, Optional[datetime]]:
        """Build the v2 author profile payload with article stats."""
        from hq.sql_models.content import Content
        from hq.sql_models.core import Author
        from sqlalchemy import func

        site = self.resolve_site_any(site_id)
        with session_scope() as session:
            author = (
                session.query(Author)
                .filter(Author.site_id == site.id, Author.slug == slug, Author.is_active.is_(True))
                .one_or_none()
            )
            if author is None:
                raise APIError("resource_not_found", "Author not found", 404, {"slug": slug})
            settings = self._effective_settings_v2(session, site.id)
            cdn = settings.get("cdn_base_url")
            count, latest = (
                session.query(func.count(Content.id), func.max(Content.published_at))
                .filter(
                    Content.site_id == site.id,
                    Content.author_id == author.id,
                    Content.status == _PUBLISHED,
                )
                .one()
            )
            author_slice = v2_common.author_slice(author, cdn)
            canonical = f"/author/{author.slug}"
            seo = v2_common.build_author_seo(author_slice, canonical)
            stats = v2_common.author_stats(int(count or 0), latest)
        profile = AuthorProfile(
            author=author_slice,
            seo=seo,
            stats=stats,
            meta=v2_common.meta_envelope("authors", latest),
        )
        return profile, latest

    def list_items(
        self,
        site_id: str,
        kind: str,
        params: Dict[str, Any],
        language: Optional[str] = None,
        limit: int = 20,
        cursor: Optional[str] = None,
    ) -> Tuple[ListResponse, Optional[datetime]]:
        """Return metadata-only content items for a list ``kind``, cursor-paginated."""
        from hq.sql_models.content import Content

        site = self.resolve_site_any(site_id)
        offset = cursor_offset(cursor)
        with session_scope() as session:
            settings = self._effective_settings_v2(session, site.id)
            cdn = settings.get("cdn_base_url")
            query = session.query(Content).filter(
                Content.site_id == site.id,
                Content.status == _PUBLISHED,
                Content.is_active.is_(True),
            )
            if language:
                query = query.filter(Content.language == language)
            query = self._apply_list_kind(session, query, site, kind, params)
            rows = (
                query.order_by(Content.published_at.desc(), Content.id.desc())
                .offset(offset)
                .limit(limit + 1)
                .all()
            )
            page = rows[:limit]
            authors = self._authors_light(session, [r.author_id for r in page])
            primary_cats = self._primary_categories(session, [r.id for r in page])
            items = [
                v2_common.list_item_from_fields(
                    id=r.id,
                    slug=r.slug,
                    title=r.title,
                    description=r.description,
                    featured_image=r.featured_image,
                    **v2_common.featured_image_variants(r),
                    author_name=authors[r.author_id][0] if r.author_id in authors else None,
                    author_image=authors[r.author_id][1] if r.author_id in authors else None,
                    category_slug=primary_cats.get(r.id),
                    published_at=r.published_at,
                    cdn_base_url=cdn,
                )
                for r in page
            ]
        next_cursor = offset_page_cursor(offset, limit, len(rows))
        last_mod = max((r.published_at for r in page if r.published_at), default=None)
        response = ListResponse(
            items=items, next_cursor=next_cursor, total=None, meta=v2_common.meta_envelope("lists", last_mod)
        )
        return response, last_mod

    def _apply_list_kind(self, session, query, site, kind, params):
        from hq.sql_models.content import Content
        from hq.sql_models.core import Author
        from hq.sql_models.taxonomy import ContentCategory

        if kind in ("homepage",):
            return query.filter(Content.show_in_homepage.is_(True))
        if kind == "featured":
            return query.filter(Content.is_featured.is_(True))
        if kind == "trending":
            return query.filter(Content.is_trending.is_(True))
        if kind == "recent":
            return query
        if kind == "category":
            category = params.get("category")
            if not category:
                raise APIError("validation_error", "category is required for kind=category", 400)
            content_ids = [
                row[0]
                for row in session.query(ContentCategory.content_id)
                .filter(ContentCategory.category == category)
                .all()
            ]
            return query.filter(Content.id.in_(content_ids or [-1]))
        if kind == "author":
            author_slug = params.get("author_slug")
            if not author_slug:
                raise APIError("validation_error", "author_slug is required for kind=author", 400)
            author = (
                session.query(Author)
                .filter(Author.site_id == site.id, Author.slug == author_slug)
                .one_or_none()
            )
            return query.filter(Content.author_id == (author.id if author else -1))
        if kind == "related":
            related_to = params.get("related_to")
            if not related_to:
                raise APIError("validation_error", "related_to is required for kind=related", 400)
            base = (
                session.query(Content)
                .filter(Content.site_id == site.id, Content.slug == related_to)
                .one_or_none()
            )
            if base is None:
                return query.filter(Content.id == -1)
            cats = [
                row[0]
                for row in session.query(ContentCategory.category)
                .filter(ContentCategory.content_id == base.id)
                .all()
            ]
            ids = [
                row[0]
                for row in session.query(ContentCategory.content_id)
                .filter(ContentCategory.category.in_(cats or [""]), ContentCategory.content_id != base.id)
                .distinct()
                .all()
            ]
            return query.filter(Content.id.in_(ids or [-1]))
        if kind == "search":
            search = params.get("q")
            if not search:
                raise APIError("validation_error", "q is required for kind=search", 400)
            like = f"%{search}%"
            return query.filter((Content.title.ilike(like)) | (Content.description.ilike(like)))
        raise APIError("validation_error", f"Unknown list kind: {kind}", 400, {"kind": kind})

    def _primary_categories(self, session, content_ids) -> Dict[int, str]:
        from hq.sql_models.taxonomy import ContentCategory

        if not content_ids:
            return {}
        rows = (
            session.query(ContentCategory.content_id, ContentCategory.category)
            .filter(ContentCategory.content_id.in_(content_ids))
            .order_by(ContentCategory.content_id, ContentCategory.category)
            .all()
        )
        primary: Dict[int, str] = {}
        for content_id, category in rows:
            primary.setdefault(content_id, category)
        return primary

    def build_sitemap(
        self,
        site_id: str,
        since: Optional[str] = None,
        limit: int = 5000,
        cursor: Optional[str] = None,
    ) -> Tuple[SitemapResponse, Optional[datetime]]:
        """Build the v2 sitemap (flagged content + static pages), cursor-paginated."""
        from hq.sql_models.content import Content
        from hq.sql_models.lookup import TemplateElement

        site = self.resolve_site_any(site_id)
        since_dt = v2_common.parse_since(since)
        offset = cursor_offset(cursor)
        with session_scope() as session:
            settings = self._effective_settings_v2(session, site.id)
            query = session.query(Content).filter(
                Content.site_id == site.id,
                Content.status == _PUBLISHED,
                Content.is_active.is_(True),
                (Content.is_featured.is_(True)) | (Content.show_in_homepage.is_(True)),
            )
            if since_dt is not None:
                query = query.filter(Content.published_at >= since_dt)
            rows = (
                query.order_by(Content.published_at.desc(), Content.id.desc())
                .offset(offset)
                .limit(limit + 1)
                .all()
            )
            primary_cats = self._primary_categories(session, [r.id for r in rows[:limit]])
            page = rows[:limit]
            entries = [
                v2_common.sitemap_entry(
                    loc=v2_common.compose_loc(primary_cats.get(r.id), r.slug),
                    lastmod=r.published_at,
                    changefreq="daily",
                    priority=0.8,
                )
                for r in page
            ]
            static_entries = []
            if offset == 0:
                navigation = settings.get("navigation") or []
                nav_urls = v2_common.navigation_urls(navigation)
                page_element_names = {
                    row.name
                    for row in session.query(TemplateElement)
                    .filter(TemplateElement.type == "page", TemplateElement.is_active.is_(True))
                    .all()
                }
                static_entries = v2_common.static_sitemap_entries(page_element_names, nav_urls)
        next_cursor = offset_page_cursor(offset, limit, len(rows))
        all_entries = static_entries + entries
        last_mod = max((r.published_at for r in page if r.published_at), default=None)
        response = SitemapResponse(
            items=all_entries,
            next_cursor=next_cursor,
            total=None,
            meta=v2_common.meta_envelope("sitemap", last_mod),
        )
        return response, last_mod

resolve_site(site_acronym)

Return an active site by acronym or domain.

Source code in src/cms_api/public/store/sql.py

def resolve_site(self, site_acronym: str) -> SiteOut:
    """Return an active site by acronym or domain."""
    from hq.sql_models.core import Site

    with session_scope() as session:
        row = (
            session.query(Site)
            .filter((Site.acronym == site_acronym) | (Site.domain == site_acronym))
            .one_or_none()
        )
    if not row or not row.is_active:
        raise APIError("resource_not_found", "Site not found", 404, {"site_acronym": site_acronym})
    return SiteOut(
        id=row.id,
        acronym=row.acronym,
        name=row.name,
        tagline=row.tagline,
        description=row.description,
        domain=row.domain,
        default_language=row.default_language,
        meta_title=row.meta_title,
        meta_description=row.meta_description,
        template_id=row.template_id,
        is_active=bool(row.is_active),
    )

resolve_site_any(site)

Resolve a site by numeric id first, then acronym/domain. Used by v2.

Source code in src/cms_api/public/store/sql.py

def resolve_site_any(self, site: str) -> SiteOut:
    """Resolve a site by numeric id first, then acronym/domain. Used by v2."""
    from hq.sql_models.core import Site

    with session_scope() as session:
        row = None
        try:
            site_id = int(site)
        except (TypeError, ValueError):
            site_id = None
        if site_id is not None:
            row = session.get(Site, site_id)
        if row is None:
            row = (
                session.query(Site)
                .filter((Site.acronym == site) | (Site.domain == site))
                .one_or_none()
            )
    if not row or not row.is_active:
        raise APIError("resource_not_found", "Site not found", 404, {"site": site})
    return SiteOut(
        id=row.id,
        acronym=row.acronym,
        name=row.name,
        tagline=row.tagline,
        description=row.description,
        domain=row.domain,
        default_language=row.default_language,
        meta_title=row.meta_title,
        meta_description=row.meta_description,
        template_id=row.template_id,
        is_active=bool(row.is_active),
    )

get_public_config(site_acronym)

Return public site configuration.

Source code in src/cms_api/public/store/sql.py

def get_public_config(self, site_acronym: str) -> Tuple[PublicSiteConfig, Optional[datetime]]:
    """Return public site configuration."""
    site = self.resolve_site(site_acronym)
    with session_scope() as session:
        settings = self._effective_settings(session, site.id)
    config = PublicSiteConfig(
        site=self._site_summary(site),
        default_meta_title=settings.get("default_meta_title"),
        default_meta_description=settings.get("default_meta_description"),
        default_og_image=absolutize_media_url(settings.get("default_og_image"), settings.get("cdn_base_url")),
        default_favicon=absolutize_media_url(settings.get("default_favicon"), settings.get("cdn_base_url")),
        cdn_base_url=settings.get("cdn_base_url"),
        maintenance_mode=bool(settings.get("maintenance_mode")),
        footer_text=settings.get("footer_text"),
        navigation=settings.get("navigation"),
        icon_url=absolutize_media_url(settings.get("icon_url"), settings.get("cdn_base_url")),
    )
    return config, datetime.now(timezone.utc)

get_published_content(site_acronym, slug, language)

Return a published content document.

Source code in src/cms_api/public/store/sql.py

def get_published_content(
    self, site_acronym: str, slug: str, language: Optional[str]
) -> Tuple[PublicContentOut, Optional[datetime]]:
    """Return a published content document."""
    site = self.resolve_site(site_acronym)
    with session_scope() as session:
        row = self._fetch_published_content(session, site.id, slug, language)
        settings = self._effective_settings(session, site.id)
        return self._content_public(row, settings.get("cdn_base_url")), row.published_at

get_seo(site_acronym, slug, language)

Return SEO metadata for a published slug.

Source code in src/cms_api/public/store/sql.py

def get_seo(self, site_acronym: str, slug: str, language: Optional[str]) -> Tuple[SeoOut, Optional[datetime]]:
    """Return SEO metadata for a published slug."""
    site = self.resolve_site(site_acronym)
    with session_scope() as session:
        row = self._fetch_published_content(session, site.id, slug, language)
        settings = self._effective_settings(session, site.id)
        return self._seo_out(row, settings.get("cdn_base_url")), row.published_at

build_page_payload(site_acronym, slug, language)

Build the full page render payload.

Source code in src/cms_api/public/store/sql.py

def build_page_payload(
    self, site_acronym: str, slug: str, language: Optional[str]
) -> Tuple[PageRenderOut, Optional[datetime]]:
    """Build the full page render payload."""
    site = self.resolve_site(site_acronym)
    with session_scope() as session:
        row = self._fetch_published_content(session, site.id, slug, language)
        settings = self._effective_settings(session, site.id)
        cdn = settings.get("cdn_base_url")
        site_summary = self._site_summary(site)
        content = self._content_public(row, cdn)
        author = self._author_public(session, row.author_id, cdn)
        seo = self._seo_out(row, cdn)
        blocks = self._resolved_blocks(session, site)
        data = {
            "site": site_summary.model_dump(),
            "content": content.model_dump(),
            "author": author.model_dump(),
        }
        payload = PageRenderOut(
            site=site_summary,
            content=content,
            author=author,
            template={"blocks": blocks},
            seo=seo,
            data=data,
            meta=PageMeta(bindings=collect_bindings(data)),
        )
        return payload, row.published_at

list_published_content(site_acronym, content_type, featured, homepage, language, limit, cursor)

Return a cursor-paginated list of published content.

Source code in src/cms_api/public/store/sql.py

def list_published_content(
    self,
    site_acronym: str,
    content_type: Optional[str],
    featured: Optional[bool],
    homepage: Optional[bool],
    language: Optional[str],
    limit: int,
    cursor: Optional[str],
) -> Tuple[List[PublicContentOut], Optional[str], Optional[datetime]]:
    """Return a cursor-paginated list of published content."""
    from hq.sql_models.content import Content

    site = self.resolve_site(site_acronym)
    offset = cursor_offset(cursor)
    with session_scope() as session:
        query = session.query(Content).filter(
            Content.site_id == site.id,
            Content.status == _PUBLISHED,
            Content.is_active.is_(True),
        )
        if content_type:
            query = query.filter(Content.content_type == content_type)
        if featured is not None:
            query = query.filter(Content.is_featured == featured)
        if homepage is not None:
            query = query.filter(Content.show_in_homepage == homepage)
        if language:
            query = query.filter(Content.language == language)
        rows = query.order_by(Content.id).offset(offset).limit(limit + 1).all()
        settings = self._effective_settings(session, site.id)
        cdn = settings.get("cdn_base_url")
    next_cursor = offset_page_cursor(offset, limit, len(rows))
    page = rows[:limit]
    items = [self._content_public(row, cdn) for row in page]
    last_mod = max((row.published_at for row in page if row.published_at), default=None)
    return items, next_cursor, last_mod

list_site_categories(site_acronym)

List enabled categories for a site.

Source code in src/cms_api/public/store/sql.py

def list_site_categories(self, site_acronym: str) -> Tuple[List[SiteCategoryPublic], Optional[datetime]]:
    """List enabled categories for a site."""
    from hq.sql_models.taxonomy import SiteCategory

    site = self.resolve_site(site_acronym)
    with session_scope() as session:
        rows = (
            session.query(SiteCategory)
            .filter(SiteCategory.site_id == site.id)
            .order_by(SiteCategory.sort_order, SiteCategory.category)
            .all()
        )
    return [
        SiteCategoryPublic(id=row.id, category=row.category, sort_order=row.sort_order or 0) for row in rows
    ], datetime.now(timezone.utc)

get_template_css(slug)

Return (css, css_version) for a template by slug (for the CSS endpoint).

Source code in src/cms_api/public/store/sql.py

def get_template_css(self, slug: str) -> Tuple[str, Optional[str]]:
    """Return ``(css, css_version)`` for a template by slug (for the CSS endpoint)."""
    from hq.sql_models.template import Template

    with session_scope() as session:
        row = (
            session.query(Template)
            .filter(Template.slug == slug, Template.is_active.is_(True))
            .one_or_none()
        )
        if row is None:
            raise APIError("resource_not_found", "Template not found", 404, {"slug": slug})
        # Read inside the session (avoid DetachedInstanceError) + getattr-defensive.
        return (getattr(row, "css", None) or "", getattr(row, "css_version", None))

build_settings_blob(site, language=None)

Build the v2 settings bundle (identity + merged settings + templates).

Source code in src/cms_api/public/store/sql.py

def build_settings_blob(self, site: str, language: Optional[str] = None) -> Tuple[SettingsBundle, Optional[datetime]]:
    """Build the v2 settings bundle (identity + merged settings + templates)."""
    resolved_site = self.resolve_site_any(site)
    with session_scope() as session:
        settings = self._effective_settings_v2(session, resolved_site.id)
        categories = self._site_categories_light(session, resolved_site.id)
        tags = self._site_tags_light(session, resolved_site.id)
        templates = self._resolved_templates_map(session, resolved_site)
        active_template = self._active_template(session, resolved_site)
        page_types = self._resolved_page_types(session, resolved_site)
        authors = self._site_authors_light(session, resolved_site.id)
    last_modified = settings.get("_settings_updated_at")
    bundle = v2_common.build_settings_blob(
        site=resolved_site,
        settings=settings,
        categories=categories,
        tags=tags,
        templates=templates,
        active_template=active_template,
        feature_flags={},
        last_modified=last_modified,
        page_types=page_types,
        authors=authors,
    )
    return bundle, last_modified

get_article_page(site_id, slug, language=None, version=None, related_limit=6)

Build the v2 article page payload, optionally from content_history.

Source code in src/cms_api/public/store/sql.py

def get_article_page(
    self,
    site_id: str,
    slug: str,
    language: Optional[str] = None,
    version: Optional[int] = None,
    related_limit: int = 6,
) -> Tuple[ArticlePage, Optional[datetime]]:
    """Build the v2 article page payload, optionally from content_history."""
    from hq.sql_models.core import Author

    site = self.resolve_site_any(site_id)
    with session_scope() as session:
        live_row = self._content_by_slug(session, site.id, slug, language)
        if live_row is None:
            raise APIError("resource_not_found", "Content not found", 404, {"slug": slug})
        snapshot_changed_at: Optional[datetime] = None
        if version:
            row, snapshot_changed_at = self._history_row(session, live_row.id, version, live_row)
        else:
            row = live_row
        settings = self._effective_settings_v2(session, site.id)
        cdn = settings.get("cdn_base_url")
        categories, tags = self._content_taxonomy(session, live_row.id)
        author = session.get(Author, row.author_id)
        if author is None:
            raise APIError("resource_not_found", "Author not found", 404, {"author_id": row.author_id})
        content = v2_common.build_article_content(row, categories, tags, cdn)
        author_slice = v2_common.author_slice(author, cdn)
        seo = v2_common.build_article_seo(row, author.name, cdn)
        breadcrumbs = v2_common.build_breadcrumbs(
            site.name, categories[0] if categories else None, content.title
        )
        related = self._related_articles(session, site.id, live_row.id, categories, related_limit, cdn)
    # On a versioned read the snapshot's own change timestamp is the resource's
    # last-modified; fall back to the live row only when changed_at is null.
    if version:
        last_modified = snapshot_changed_at or content.updated_at or content.published_at
    else:
        last_modified = content.updated_at or content.published_at
    page = ArticlePage(
        content=content,
        author=author_slice,
        seo=seo,
        breadcrumbs=breadcrumbs,
        related=related,
        meta=v2_common.meta_envelope("pages", last_modified),
    )
    return page, last_modified

get_author_profile(site_id, slug)

Build the v2 author profile payload with article stats.

Source code in src/cms_api/public/store/sql.py

def get_author_profile(self, site_id: str, slug: str) -> Tuple[AuthorProfile, Optional[datetime]]:
    """Build the v2 author profile payload with article stats."""
    from hq.sql_models.content import Content
    from hq.sql_models.core import Author
    from sqlalchemy import func

    site = self.resolve_site_any(site_id)
    with session_scope() as session:
        author = (
            session.query(Author)
            .filter(Author.site_id == site.id, Author.slug == slug, Author.is_active.is_(True))
            .one_or_none()
        )
        if author is None:
            raise APIError("resource_not_found", "Author not found", 404, {"slug": slug})
        settings = self._effective_settings_v2(session, site.id)
        cdn = settings.get("cdn_base_url")
        count, latest = (
            session.query(func.count(Content.id), func.max(Content.published_at))
            .filter(
                Content.site_id == site.id,
                Content.author_id == author.id,
                Content.status == _PUBLISHED,
            )
            .one()
        )
        author_slice = v2_common.author_slice(author, cdn)
        canonical = f"/author/{author.slug}"
        seo = v2_common.build_author_seo(author_slice, canonical)
        stats = v2_common.author_stats(int(count or 0), latest)
    profile = AuthorProfile(
        author=author_slice,
        seo=seo,
        stats=stats,
        meta=v2_common.meta_envelope("authors", latest),
    )
    return profile, latest

list_items(site_id, kind, params, language=None, limit=20, cursor=None)

Return metadata-only content items for a list kind, cursor-paginated.

Source code in src/cms_api/public/store/sql.py

def list_items(
    self,
    site_id: str,
    kind: str,
    params: Dict[str, Any],
    language: Optional[str] = None,
    limit: int = 20,
    cursor: Optional[str] = None,
) -> Tuple[ListResponse, Optional[datetime]]:
    """Return metadata-only content items for a list ``kind``, cursor-paginated."""
    from hq.sql_models.content import Content

    site = self.resolve_site_any(site_id)
    offset = cursor_offset(cursor)
    with session_scope() as session:
        settings = self._effective_settings_v2(session, site.id)
        cdn = settings.get("cdn_base_url")
        query = session.query(Content).filter(
            Content.site_id == site.id,
            Content.status == _PUBLISHED,
            Content.is_active.is_(True),
        )
        if language:
            query = query.filter(Content.language == language)
        query = self._apply_list_kind(session, query, site, kind, params)
        rows = (
            query.order_by(Content.published_at.desc(), Content.id.desc())
            .offset(offset)
            .limit(limit + 1)
            .all()
        )
        page = rows[:limit]
        authors = self._authors_light(session, [r.author_id for r in page])
        primary_cats = self._primary_categories(session, [r.id for r in page])
        items = [
            v2_common.list_item_from_fields(
                id=r.id,
                slug=r.slug,
                title=r.title,
                description=r.description,
                featured_image=r.featured_image,
                **v2_common.featured_image_variants(r),
                author_name=authors[r.author_id][0] if r.author_id in authors else None,
                author_image=authors[r.author_id][1] if r.author_id in authors else None,
                category_slug=primary_cats.get(r.id),
                published_at=r.published_at,
                cdn_base_url=cdn,
            )
            for r in page
        ]
    next_cursor = offset_page_cursor(offset, limit, len(rows))
    last_mod = max((r.published_at for r in page if r.published_at), default=None)
    response = ListResponse(
        items=items, next_cursor=next_cursor, total=None, meta=v2_common.meta_envelope("lists", last_mod)
    )
    return response, last_mod

build_sitemap(site_id, since=None, limit=5000, cursor=None)

Build the v2 sitemap (flagged content + static pages), cursor-paginated.

Source code in src/cms_api/public/store/sql.py

def build_sitemap(
    self,
    site_id: str,
    since: Optional[str] = None,
    limit: int = 5000,
    cursor: Optional[str] = None,
) -> Tuple[SitemapResponse, Optional[datetime]]:
    """Build the v2 sitemap (flagged content + static pages), cursor-paginated."""
    from hq.sql_models.content import Content
    from hq.sql_models.lookup import TemplateElement

    site = self.resolve_site_any(site_id)
    since_dt = v2_common.parse_since(since)
    offset = cursor_offset(cursor)
    with session_scope() as session:
        settings = self._effective_settings_v2(session, site.id)
        query = session.query(Content).filter(
            Content.site_id == site.id,
            Content.status == _PUBLISHED,
            Content.is_active.is_(True),
            (Content.is_featured.is_(True)) | (Content.show_in_homepage.is_(True)),
        )
        if since_dt is not None:
            query = query.filter(Content.published_at >= since_dt)
        rows = (
            query.order_by(Content.published_at.desc(), Content.id.desc())
            .offset(offset)
            .limit(limit + 1)
            .all()
        )
        primary_cats = self._primary_categories(session, [r.id for r in rows[:limit]])
        page = rows[:limit]
        entries = [
            v2_common.sitemap_entry(
                loc=v2_common.compose_loc(primary_cats.get(r.id), r.slug),
                lastmod=r.published_at,
                changefreq="daily",
                priority=0.8,
            )
            for r in page
        ]
        static_entries = []
        if offset == 0:
            navigation = settings.get("navigation") or []
            nav_urls = v2_common.navigation_urls(navigation)
            page_element_names = {
                row.name
                for row in session.query(TemplateElement)
                .filter(TemplateElement.type == "page", TemplateElement.is_active.is_(True))
                .all()
            }
            static_entries = v2_common.static_sitemap_entries(page_element_names, nav_urls)
    next_cursor = offset_page_cursor(offset, limit, len(rows))
    all_entries = static_entries + entries
    last_mod = max((r.published_at for r in page if r.published_at), default=None)
    response = SitemapResponse(
        items=all_entries,
        next_cursor=next_cursor,
        total=None,
        meta=v2_common.meta_envelope("sitemap", last_mod),
    )
    return response, last_mod

Memory store

memory

In-memory public store for unit tests.

MemoryPublicStore

Read-only in-memory persistence for public routes in tests.

Source code in src/cms_api/public/store/memory.py

class MemoryPublicStore:
    """Read-only in-memory persistence for public routes in tests."""

    def __init__(self) -> None:
        """Initialize empty collections."""
        self.sites: Dict[int, SiteOut] = {}
        self.settings: Dict[int, SiteSettingsOut] = {}
        self.authors: Dict[int, AuthorOut] = {}
        self.content: Dict[int, ContentOut] = {}
        self.templates: Dict[int, Any] = {}
        self.template_blocks: Dict[int, TemplateBlockOut] = {}
        self.block_overrides: Dict[int, SiteBlockOverrideOut] = {}
        # v2 page descriptors (AISD P1.b): template_id -> list of raw descriptor
        # dicts (same shape as page_types rows), validated by the shared shaper.
        self.page_types: Dict[int, List[Dict[str, Any]]] = {}
        self.site_categories: Dict[int, SiteCategoryOut] = {}
        # v2 lookup support (AISD-116): name->id maps and template_elements.
        self.categories: Dict[str, int] = {}
        self.tags: Dict[str, int] = {}
        self.template_elements: Dict[str, Dict[str, Any]] = {}
        self.content_history: Dict[int, List[ContentOut]] = {}
        # Parallel to content_history: the changed_at timestamp per snapshot
        # (mirrors ContentHistory.changed_at). Absent entries default to None.
        self.content_history_changed_at: Dict[int, List[Optional[datetime]]] = {}

    def resolve_site(self, site_acronym: str) -> SiteOut:
        """Return an active site by acronym or domain."""
        for site in self.sites.values():
            if site.acronym == site_acronym or site.domain == site_acronym:
                if not site.is_active:
                    raise APIError("resource_not_found", "Site not found", 404, {"site_acronym": site_acronym})
                return site
        raise APIError("resource_not_found", "Site not found", 404, {"site_acronym": site_acronym})

    def _settings_row(self, site_id: int) -> SiteSettingsOut:
        if site_id not in self.settings:
            self.settings[site_id] = SiteSettingsOut(site_id=site_id)
        return self.settings[site_id]

    def _effective_settings(self, site_id: int) -> SiteSettingsOut:
        global_row = self._settings_row(GLOBAL_SETTINGS_SITE_ID)
        site_row = self._settings_row(site_id)
        merged: Dict[str, Any] = {"site_id": site_id}
        for field in _SETTINGS_FIELDS:
            site_val = getattr(site_row, field, None)
            global_val = getattr(global_row, field, None)
            merged[field] = site_val if site_val is not None else global_val
        return SiteSettingsOut(**merged)

    def _site_summary(self, site: SiteOut) -> PublicSiteSummary:
        return PublicSiteSummary(
            id=site.id,
            acronym=site.acronym,
            name=site.name,
            domain=site.domain,
            default_language=site.default_language,
            tagline=site.tagline,
        )

    def get_public_config(self, site_acronym: str) -> Tuple[PublicSiteConfig, Optional[datetime]]:
        """Return public site configuration."""
        site = self.resolve_site(site_acronym)
        settings = self._effective_settings(site.id)
        config = PublicSiteConfig(
            site=self._site_summary(site),
            default_meta_title=settings.default_meta_title,
            default_meta_description=settings.default_meta_description,
            default_og_image=absolutize_media_url(settings.default_og_image, settings.cdn_base_url),
            default_favicon=absolutize_media_url(settings.default_favicon, settings.cdn_base_url),
            cdn_base_url=settings.cdn_base_url,
            maintenance_mode=bool(settings.maintenance_mode),
            footer_text=settings.footer_text,
            navigation=settings.navigation,
            icon_url=absolutize_media_url(settings.icon_url, settings.cdn_base_url),
        )
        return config, datetime.now(timezone.utc)

    def _published_content(self, site_id: int, slug: str, language: Optional[str]) -> ContentOut:
        for item in self.content.values():
            if item.site_id != site_id or item.slug != slug:
                continue
            if item.status != _PUBLISHED or not item.is_active:
                raise APIError("resource_not_found", "Content not found", 404, {"slug": slug})
            if language and item.language != language:
                continue
            return item
        raise APIError("resource_not_found", "Content not found", 404, {"slug": slug})

    def _to_public_content(self, item: ContentOut, cdn_base_url: Optional[str]) -> PublicContentOut:
        return PublicContentOut(
            id=item.id,
            site_id=item.site_id,
            content_type=item.content_type,
            language=item.language,
            slug=item.slug,
            title=item.title,
            description=item.description,
            author_id=item.author_id,
            featured_image=absolutize_media_url(item.featured_image, cdn_base_url),
            show_in_homepage=item.show_in_homepage,
            is_featured=item.is_featured,
            is_trending=item.is_trending,
            content_blocks=item.content_blocks,
            settings=item.settings,
            published_at=item.published_at,
            categories=list(item.categories),
            tags=list(item.tags),
        )

    def _author_public(self, author_id: int, cdn_base_url: Optional[str]) -> PublicAuthorOut:
        author = self.authors.get(author_id)
        if not author or not author.is_active:
            raise APIError("resource_not_found", "Author not found", 404, {"author_id": author_id})
        return PublicAuthorOut(
            id=author.id,
            slug=author.slug,
            name=author.name,
            initials=author.initials,
            image=absolutize_media_url(author.image, cdn_base_url),
            bio=author.bio,
        )

    def _seo_out(self, item: ContentOut, cdn_base_url: Optional[str]) -> SeoOut:
        return SeoOut(
            slug=item.slug,
            seo_title=item.seo_title or item.title,
            seo_description=item.seo_description or item.description,
            og_title=item.og_title,
            og_description=item.og_description,
            og_image=absolutize_media_url(item.og_image or item.featured_image, cdn_base_url),
            canonical_url=item.canonical_url,
            noindex=item.noindex,
            schema_type=item.schema_type,
        )

    def _resolved_blocks(self, site: SiteOut) -> List[TemplateBlockPublic]:
        if not site.template_id:
            return []
        overrides = {
            row.element_type: row
            for row in self.block_overrides.values()
            if row.site_id == site.id
        }
        blocks: List[TemplateBlockPublic] = []
        for row in self.template_blocks.values():
            if row.template_id != site.template_id:
                continue
            override = overrides.get(row.element_type)
            blocks.append(
                TemplateBlockPublic(
                    element_type=row.element_type,
                    html_content=override.html_content if override else row.html_content,
                    settings=override.settings if override and override.settings is not None else row.settings,
                )
            )
        return blocks

    def get_published_content(
        self, site_acronym: str, slug: str, language: Optional[str]
    ) -> Tuple[PublicContentOut, Optional[datetime]]:
        """Return a published content document."""
        site = self.resolve_site(site_acronym)
        item = self._published_content(site.id, slug, language)
        settings = self._effective_settings(site.id)
        return self._to_public_content(item, settings.cdn_base_url), item.published_at

    def get_seo(self, site_acronym: str, slug: str, language: Optional[str]) -> Tuple[SeoOut, Optional[datetime]]:
        """Return SEO metadata for a published slug."""
        site = self.resolve_site(site_acronym)
        item = self._published_content(site.id, slug, language)
        settings = self._effective_settings(site.id)
        return self._seo_out(item, settings.cdn_base_url), item.published_at

    def build_page_payload(
        self, site_acronym: str, slug: str, language: Optional[str]
    ) -> Tuple[PageRenderOut, Optional[datetime]]:
        """Build the full page render payload."""
        site = self.resolve_site(site_acronym)
        item = self._published_content(site.id, slug, language)
        settings = self._effective_settings(site.id)
        cdn = settings.cdn_base_url
        site_summary = self._site_summary(site)
        content = self._to_public_content(item, cdn)
        author = self._author_public(item.author_id, cdn)
        seo = self._seo_out(item, cdn)
        blocks = self._resolved_blocks(site)
        data = {
            "site": site_summary.model_dump(),
            "content": content.model_dump(),
            "author": author.model_dump(),
        }
        payload = PageRenderOut(
            site=site_summary,
            content=content,
            author=author,
            template={"blocks": blocks},
            seo=seo,
            data=data,
            meta=PageMeta(bindings=collect_bindings(data)),
        )
        return payload, item.published_at

    def list_published_content(
        self,
        site_acronym: str,
        content_type: Optional[str],
        featured: Optional[bool],
        homepage: Optional[bool],
        language: Optional[str],
        limit: int,
        cursor: Optional[str],
    ) -> Tuple[List[PublicContentOut], Optional[str], Optional[datetime]]:
        """Return a cursor-paginated list of published content."""
        site = self.resolve_site(site_acronym)
        settings = self._effective_settings(site.id)
        ids = sorted(
            item.id
            for item in self.content.values()
            if item.site_id == site.id
            and item.status == _PUBLISHED
            and item.is_active
            and (content_type is None or item.content_type == content_type)
            and (featured is None or item.is_featured == featured)
            and (homepage is None or item.show_in_homepage == homepage)
            and (language is None or item.language == language)
        )
        page_ids, next_cursor = paginate_ids(ids, limit, cursor)
        items = [self._to_public_content(self.content[i], settings.cdn_base_url) for i in page_ids]
        last_mod = None
        for item_id in page_ids:
            published_at = self.content[item_id].published_at
            if published_at and (last_mod is None or published_at > last_mod):
                last_mod = published_at
        return items, next_cursor, last_mod

    def list_site_categories(self, site_acronym: str) -> Tuple[List[SiteCategoryPublic], Optional[datetime]]:
        """List enabled categories for a site."""
        site = self.resolve_site(site_acronym)
        rows = sorted(
            (
                SiteCategoryPublic(id=row.id, category=row.category, sort_order=row.sort_order)
                for row in self.site_categories.values()
                if row.site_id == site.id
            ),
            key=lambda row: (row.sort_order, row.category),
        )
        return rows, datetime.now(timezone.utc)

    # --- v2 worker-facing surface (AISD-116) -------------------------------

    def resolve_site_any(self, site: str) -> SiteOut:
        """Resolve a site by numeric id first, then acronym/domain. Used by v2."""
        try:
            site_id = int(site)
        except (TypeError, ValueError):
            site_id = None
        if site_id is not None and site_id in self.sites:
            row = self.sites[site_id]
            if not row.is_active:
                raise APIError("resource_not_found", "Site not found", 404, {"site": site})
            return row
        for row in self.sites.values():
            if row.acronym == site or row.domain == site:
                if not row.is_active:
                    raise APIError("resource_not_found", "Site not found", 404, {"site": site})
                return row
        raise APIError("resource_not_found", "Site not found", 404, {"site": site})

    def _effective_settings_v2(self, site_id: int) -> Dict[str, Any]:
        global_row = self._settings_row(GLOBAL_SETTINGS_SITE_ID)
        site_row = self._settings_row(site_id)
        merged: Dict[str, Any] = {"site_id": site_id}
        for field in v2_common.SETTINGS_FIELDS:
            site_val = getattr(site_row, field, None)
            global_val = getattr(global_row, field, None)
            merged[field] = site_val if site_val is not None else global_val
        return merged

    def _categories_light_v2(self, site_id: int) -> List[Tuple[int, str, str, int]]:
        result: List[Tuple[int, str, str, int]] = []
        rows = sorted(
            (row for row in self.site_categories.values() if row.site_id == site_id),
            key=lambda row: (row.sort_order or 0, row.category),
        )
        for row in rows:
            cat_id = self.categories.get(row.category, row.id)
            result.append((cat_id, row.category, row.category, row.sort_order or 0))
        return result

    def _tags_light_v2(self, site_id: int) -> List[Tuple[int, str, str]]:
        names = sorted(
            {
                tag
                for item in self.content.values()
                if item.site_id == site_id and item.status == _PUBLISHED
                for tag in item.tags
            }
        )
        return [(self.tags.get(name, idx + 1), name, name) for idx, name in enumerate(names)]

    def _resolved_templates_map_v2(self, site: SiteOut) -> Dict[str, List[Tuple[Optional[str], Optional[Any]]]]:
        blocks = {
            row.element_type: row
            for row in self.template_blocks.values()
            if site.template_id and row.template_id == site.template_id
        }
        overrides = {
            row.element_type: row for row in self.block_overrides.values() if row.site_id == site.id
        }
        resolved: Dict[str, List[Tuple[Optional[str], Optional[Any]]]] = {}
        element_names = set(self.template_elements) | set(blocks) | set(overrides)
        for name in sorted(element_names):
            element = self.template_elements.get(name, {})
            html = element.get("html_content")
            settings = element.get("default_settings")
            block = blocks.get(name)
            override = overrides.get(name)
            if block is not None:
                if block.html_content is not None:
                    html = block.html_content
                if block.settings is not None:
                    settings = block.settings
            if override is not None:
                if override.html_content is not None:
                    html = override.html_content
                if override.settings is not None:
                    settings = override.settings
            resolved[name] = [(html, settings)]
        return resolved

    def _active_template_v2(
        self, site: SiteOut
    ) -> Optional[Tuple[int, str, Optional[str], Optional[str]]]:
        if not site.template_id:
            return None
        template = self.templates.get(site.template_id)
        if template is None:
            return None
        return (
            template.id,
            template.slug,
            getattr(template, "version", None),
            getattr(template, "css_version", None),
        )

    def _resolved_page_types_v2(self, site: SiteOut) -> List[Dict[str, Any]]:
        """Return the active template's seeded page descriptors (P1.b)."""
        if not site.template_id:
            return []
        return list(self.page_types.get(site.template_id, []))

    def get_template_css(self, slug: str) -> Tuple[str, Optional[str]]:
        """Return ``(css, css_version)`` for a template by slug (for the CSS endpoint)."""
        for template in self.templates.values():
            if getattr(template, "slug", None) == slug:
                return (getattr(template, "css", None) or "", getattr(template, "css_version", None))
        raise APIError("resource_not_found", "Template not found", 404, {"slug": slug})

    def _authors_light_v2(self, site_id: int) -> List[Dict[str, Any]]:
        """All active authors for a site + published-article counts (for /authors)."""
        counts: Dict[int, int] = {}
        for item in self.content.values():
            if item.site_id == site_id and item.status == _PUBLISHED and item.is_active:
                counts[item.author_id] = counts.get(item.author_id, 0) + 1
        authors = [a for a in self.authors.values() if a.site_id == site_id and a.is_active]
        authors.sort(key=lambda a: a.name)
        return [
            {
                "id": a.id,
                "slug": a.slug,
                "name": a.name,
                "initials": a.initials,
                "image": a.image,
                "bio": a.bio,
                "article_count": counts.get(a.id, 0),
            }
            for a in authors
        ]

    def build_settings_blob(
        self, site: str, language: Optional[str] = None
    ) -> Tuple[SettingsBundle, Optional[datetime]]:
        """Build the v2 settings bundle (identity + merged settings + templates)."""
        resolved_site = self.resolve_site_any(site)
        settings = self._effective_settings_v2(resolved_site.id)
        bundle = v2_common.build_settings_blob(
            site=resolved_site,
            settings=settings,
            categories=self._categories_light_v2(resolved_site.id),
            tags=self._tags_light_v2(resolved_site.id),
            templates=self._resolved_templates_map_v2(resolved_site),
            active_template=self._active_template_v2(resolved_site),
            feature_flags={},
            last_modified=None,
            page_types=self._resolved_page_types_v2(resolved_site),
            authors=self._authors_light_v2(resolved_site.id),
        )
        return bundle, None

    def _published_article(self, site_id: int, slug: str, language: Optional[str]) -> ContentOut:
        for item in self.content.values():
            if item.site_id != site_id or item.slug != slug:
                continue
            # Language mismatch must skip ahead (mirrors the SQL query filter) so a
            # slug published in another language does not short-circuit to 404.
            if language and item.language != language:
                continue
            if (
                item.content_type not in v2_common.ARTICLE_CONTENT_TYPES
                or item.status != _PUBLISHED
                or not item.is_active
            ):
                raise APIError("resource_not_found", "Content not found", 404, {"slug": slug})
            return item
        raise APIError("resource_not_found", "Content not found", 404, {"slug": slug})

    def get_article_page(
        self,
        site_id: str,
        slug: str,
        language: Optional[str] = None,
        version: Optional[int] = None,
        related_limit: int = 6,
    ) -> Tuple[ArticlePage, Optional[datetime]]:
        """Build the v2 article page payload, optionally from content_history."""
        site = self.resolve_site_any(site_id)
        live = self._published_article(site.id, slug, language)
        settings = self._effective_settings_v2(site.id)
        cdn = settings.get("cdn_base_url")
        snapshot_changed_at: Optional[datetime] = None
        if version:
            row, snapshot_changed_at = self._merged_history(live, version)
        else:
            row = live
        author = self.authors.get(row.author_id)
        if author is None or not author.is_active:
            raise APIError("resource_not_found", "Author not found", 404, {"author_id": row.author_id})
        content = v2_common.build_article_content(row, row.categories, row.tags, cdn)
        author_slice = v2_common.author_slice(author, cdn)
        seo = v2_common.build_article_seo(row, author.name, cdn)
        breadcrumbs = v2_common.build_breadcrumbs(
            site.name, row.categories[0] if row.categories else None, content.title
        )
        related = self._related_v2(site.id, live, related_limit, cdn)
        # On a versioned read the snapshot's change timestamp is the resource's
        # last-modified; fall back to the live row only when it is null.
        if version:
            last_mod = snapshot_changed_at or content.updated_at or content.published_at
        else:
            last_mod = content.updated_at or content.published_at
        page = ArticlePage(
            content=content,
            author=author_slice,
            seo=seo,
            breadcrumbs=breadcrumbs,
            related=related,
            meta=v2_common.meta_envelope("pages", last_mod),
        )
        return page, last_mod

    def _merged_history(self, live: ContentOut, version: int) -> Tuple[ContentOut, Optional[datetime]]:
        """Return a (merged snapshot, changed_at) tuple for ``version``.

        Keeps the live row's identity columns and overlays only the snapshot's
        mutable business columns, mirroring the SQL store's merge policy.
        """
        history = self.content_history.get(live.id, [])
        if version < 1 or version > len(history):
            raise APIError("resource_not_found", "Content version not found", 404, {"version": version})
        snapshot = history[version - 1]
        merged = live.model_dump()
        snapshot_data = snapshot.model_dump()
        for key in v2_common.HISTORY_BUSINESS_FIELDS:
            value = snapshot_data.get(key)
            if value is not None:
                merged[key] = value
        changed_at = self.content_history_changed_at.get(live.id, [])
        snapshot_changed_at = changed_at[version - 1] if version - 1 < len(changed_at) else None
        return ContentOut(**merged), snapshot_changed_at

    def _related_v2(self, site_id: int, base: ContentOut, limit: int, cdn: Optional[str]):
        base_cats = set(base.categories)
        if not base_cats:
            return []
        candidates = [
            item
            for item in self.content.values()
            if item.site_id == site_id
            and item.id != base.id
            and item.content_type in v2_common.ARTICLE_CONTENT_TYPES
            and item.status == _PUBLISHED
            and item.is_active
            and base_cats.intersection(item.categories)
        ]
        candidates.sort(key=lambda item: (item.published_at or datetime.min.replace(tzinfo=timezone.utc), item.id), reverse=True)
        related = []
        for item in candidates[:limit]:
            author = self.authors.get(item.author_id)
            related.append(
                v2_common.related_item(
                    id=item.id,
                    slug=item.slug,
                    title=item.title,
                    featured_image=item.featured_image,
                    author_name=author.name if author else None,
                    published_at=item.published_at,
                    cdn_base_url=cdn,
                )
            )
        return related

    def get_author_profile(self, site_id: str, slug: str) -> Tuple[AuthorProfile, Optional[datetime]]:
        """Build the v2 author profile payload with article stats."""
        site = self.resolve_site_any(site_id)
        author = None
        for candidate in self.authors.values():
            if candidate.site_id == site.id and candidate.slug == slug and candidate.is_active:
                author = candidate
                break
        if author is None:
            raise APIError("resource_not_found", "Author not found", 404, {"slug": slug})
        settings = self._effective_settings_v2(site.id)
        cdn = settings.get("cdn_base_url")
        published = [
            item
            for item in self.content.values()
            if item.site_id == site.id and item.author_id == author.id and item.status == _PUBLISHED
        ]
        count = len(published)
        latest = max((item.published_at for item in published if item.published_at), default=None)
        author_slice = v2_common.author_slice(author, cdn)
        seo = v2_common.build_author_seo(author_slice, f"/author/{author.slug}")
        stats = v2_common.author_stats(count, latest)
        profile = AuthorProfile(
            author=author_slice, seo=seo, stats=stats, meta=v2_common.meta_envelope("authors", latest)
        )
        return profile, latest

    def list_items(
        self,
        site_id: str,
        kind: str,
        params: Dict[str, Any],
        language: Optional[str] = None,
        limit: int = 20,
        cursor: Optional[str] = None,
    ) -> Tuple[ListResponse, Optional[datetime]]:
        """Return metadata-only content items for a list ``kind``, cursor-paginated."""
        site = self.resolve_site_any(site_id)
        settings = self._effective_settings_v2(site.id)
        cdn = settings.get("cdn_base_url")
        matched = [
            item
            for item in self.content.values()
            if item.site_id == site.id
            and item.status == _PUBLISHED
            and item.is_active
            and (language is None or item.language == language)
            and self._matches_kind(item, kind, params, site.id)
        ]
        matched.sort(
            key=lambda item: (item.published_at or datetime.min.replace(tzinfo=timezone.utc), item.id),
            reverse=True,
        )
        ids = [item.id for item in matched]
        page_ids, next_cursor = paginate_ids(ids, limit, cursor)
        by_id = {item.id: item for item in matched}
        items = []
        for item_id in page_ids:
            item = by_id[item_id]
            author = self.authors.get(item.author_id)
            items.append(
                v2_common.list_item_from_fields(
                    id=item.id,
                    slug=item.slug,
                    title=item.title,
                    description=item.description,
                    featured_image=item.featured_image,
                    **v2_common.featured_image_variants(item),
                    author_name=author.name if author else None,
                    author_image=author.image if author else None,
                    category_slug=item.categories[0] if item.categories else None,
                    published_at=item.published_at,
                    cdn_base_url=cdn,
                )
            )
        last_mod = max((by_id[i].published_at for i in page_ids if by_id[i].published_at), default=None)
        response = ListResponse(
            items=items, next_cursor=next_cursor, total=None, meta=v2_common.meta_envelope("lists", last_mod)
        )
        return response, last_mod

    def _matches_kind(self, item: ContentOut, kind: str, params: Dict[str, Any], site_id: int) -> bool:
        if kind == "homepage":
            return bool(item.show_in_homepage)
        if kind == "featured":
            return bool(item.is_featured)
        if kind == "trending":
            return bool(item.is_trending)
        if kind == "recent":
            return True
        if kind == "category":
            category = params.get("category")
            if not category:
                raise APIError("validation_error", "category is required for kind=category", 400)
            return category in item.categories
        if kind == "author":
            author_slug = params.get("author_slug")
            if not author_slug:
                raise APIError("validation_error", "author_slug is required for kind=author", 400)
            author = next(
                (a for a in self.authors.values() if a.site_id == site_id and a.slug == author_slug), None
            )
            return author is not None and item.author_id == author.id
        if kind == "related":
            related_to = params.get("related_to")
            if not related_to:
                raise APIError("validation_error", "related_to is required for kind=related", 400)
            base = next(
                (c for c in self.content.values() if c.site_id == site_id and c.slug == related_to), None
            )
            if base is None or item.slug == related_to:
                return False
            return bool(set(base.categories).intersection(item.categories))
        if kind == "search":
            search = params.get("q")
            if not search:
                raise APIError("validation_error", "q is required for kind=search", 400)
            needle = search.lower()
            return needle in item.title.lower() or needle in (item.description or "").lower()
        raise APIError("validation_error", f"Unknown list kind: {kind}", 400, {"kind": kind})

    def build_sitemap(
        self,
        site_id: str,
        since: Optional[str] = None,
        limit: int = 5000,
        cursor: Optional[str] = None,
    ) -> Tuple[SitemapResponse, Optional[datetime]]:
        """Build the v2 sitemap (flagged content + static pages), cursor-paginated."""
        site = self.resolve_site_any(site_id)
        settings = self._effective_settings_v2(site.id)
        since_dt = v2_common.parse_since(since)
        flagged = [
            item
            for item in self.content.values()
            if item.site_id == site.id
            and item.status == _PUBLISHED
            and item.is_active
            and (item.is_featured or item.show_in_homepage)
            and (since_dt is None or (item.published_at and item.published_at >= since_dt))
        ]
        flagged.sort(
            key=lambda item: (item.published_at or datetime.min.replace(tzinfo=timezone.utc), item.id),
            reverse=True,
        )
        ids = [item.id for item in flagged]
        page_ids, next_cursor = paginate_ids(ids, limit, cursor)
        by_id = {item.id: item for item in flagged}
        entries = [
            v2_common.sitemap_entry(
                loc=v2_common.compose_loc(by_id[i].categories[0] if by_id[i].categories else None, by_id[i].slug),
                lastmod=by_id[i].published_at,
                changefreq="daily",
                priority=0.8,
            )
            for i in page_ids
        ]
        static_entries = []
        if not cursor:
            nav_urls = v2_common.navigation_urls(settings.get("navigation") or [])
            page_names = {name for name, el in self.template_elements.items() if el.get("type") == "page"}
            static_entries = v2_common.static_sitemap_entries(page_names, nav_urls)
        all_entries = static_entries + entries
        last_mod = max((by_id[i].published_at for i in page_ids if by_id[i].published_at), default=None)
        response = SitemapResponse(
            items=all_entries, next_cursor=next_cursor, total=None, meta=v2_common.meta_envelope("sitemap", last_mod)
        )
        return response, last_mod

resolve_site(site_acronym)

Return an active site by acronym or domain.

Source code in src/cms_api/public/store/memory.py

def resolve_site(self, site_acronym: str) -> SiteOut:
    """Return an active site by acronym or domain."""
    for site in self.sites.values():
        if site.acronym == site_acronym or site.domain == site_acronym:
            if not site.is_active:
                raise APIError("resource_not_found", "Site not found", 404, {"site_acronym": site_acronym})
            return site
    raise APIError("resource_not_found", "Site not found", 404, {"site_acronym": site_acronym})

get_public_config(site_acronym)

Return public site configuration.

Source code in src/cms_api/public/store/memory.py

def get_public_config(self, site_acronym: str) -> Tuple[PublicSiteConfig, Optional[datetime]]:
    """Return public site configuration."""
    site = self.resolve_site(site_acronym)
    settings = self._effective_settings(site.id)
    config = PublicSiteConfig(
        site=self._site_summary(site),
        default_meta_title=settings.default_meta_title,
        default_meta_description=settings.default_meta_description,
        default_og_image=absolutize_media_url(settings.default_og_image, settings.cdn_base_url),
        default_favicon=absolutize_media_url(settings.default_favicon, settings.cdn_base_url),
        cdn_base_url=settings.cdn_base_url,
        maintenance_mode=bool(settings.maintenance_mode),
        footer_text=settings.footer_text,
        navigation=settings.navigation,
        icon_url=absolutize_media_url(settings.icon_url, settings.cdn_base_url),
    )
    return config, datetime.now(timezone.utc)

get_published_content(site_acronym, slug, language)

Return a published content document.

Source code in src/cms_api/public/store/memory.py

def get_published_content(
    self, site_acronym: str, slug: str, language: Optional[str]
) -> Tuple[PublicContentOut, Optional[datetime]]:
    """Return a published content document."""
    site = self.resolve_site(site_acronym)
    item = self._published_content(site.id, slug, language)
    settings = self._effective_settings(site.id)
    return self._to_public_content(item, settings.cdn_base_url), item.published_at

get_seo(site_acronym, slug, language)

Return SEO metadata for a published slug.

Source code in src/cms_api/public/store/memory.py

def get_seo(self, site_acronym: str, slug: str, language: Optional[str]) -> Tuple[SeoOut, Optional[datetime]]:
    """Return SEO metadata for a published slug."""
    site = self.resolve_site(site_acronym)
    item = self._published_content(site.id, slug, language)
    settings = self._effective_settings(site.id)
    return self._seo_out(item, settings.cdn_base_url), item.published_at

build_page_payload(site_acronym, slug, language)

Build the full page render payload.

Source code in src/cms_api/public/store/memory.py

def build_page_payload(
    self, site_acronym: str, slug: str, language: Optional[str]
) -> Tuple[PageRenderOut, Optional[datetime]]:
    """Build the full page render payload."""
    site = self.resolve_site(site_acronym)
    item = self._published_content(site.id, slug, language)
    settings = self._effective_settings(site.id)
    cdn = settings.cdn_base_url
    site_summary = self._site_summary(site)
    content = self._to_public_content(item, cdn)
    author = self._author_public(item.author_id, cdn)
    seo = self._seo_out(item, cdn)
    blocks = self._resolved_blocks(site)
    data = {
        "site": site_summary.model_dump(),
        "content": content.model_dump(),
        "author": author.model_dump(),
    }
    payload = PageRenderOut(
        site=site_summary,
        content=content,
        author=author,
        template={"blocks": blocks},
        seo=seo,
        data=data,
        meta=PageMeta(bindings=collect_bindings(data)),
    )
    return payload, item.published_at

list_published_content(site_acronym, content_type, featured, homepage, language, limit, cursor)

Return a cursor-paginated list of published content.

Source code in src/cms_api/public/store/memory.py

def list_published_content(
    self,
    site_acronym: str,
    content_type: Optional[str],
    featured: Optional[bool],
    homepage: Optional[bool],
    language: Optional[str],
    limit: int,
    cursor: Optional[str],
) -> Tuple[List[PublicContentOut], Optional[str], Optional[datetime]]:
    """Return a cursor-paginated list of published content."""
    site = self.resolve_site(site_acronym)
    settings = self._effective_settings(site.id)
    ids = sorted(
        item.id
        for item in self.content.values()
        if item.site_id == site.id
        and item.status == _PUBLISHED
        and item.is_active
        and (content_type is None or item.content_type == content_type)
        and (featured is None or item.is_featured == featured)
        and (homepage is None or item.show_in_homepage == homepage)
        and (language is None or item.language == language)
    )
    page_ids, next_cursor = paginate_ids(ids, limit, cursor)
    items = [self._to_public_content(self.content[i], settings.cdn_base_url) for i in page_ids]
    last_mod = None
    for item_id in page_ids:
        published_at = self.content[item_id].published_at
        if published_at and (last_mod is None or published_at > last_mod):
            last_mod = published_at
    return items, next_cursor, last_mod

list_site_categories(site_acronym)

List enabled categories for a site.

Source code in src/cms_api/public/store/memory.py

def list_site_categories(self, site_acronym: str) -> Tuple[List[SiteCategoryPublic], Optional[datetime]]:
    """List enabled categories for a site."""
    site = self.resolve_site(site_acronym)
    rows = sorted(
        (
            SiteCategoryPublic(id=row.id, category=row.category, sort_order=row.sort_order)
            for row in self.site_categories.values()
            if row.site_id == site.id
        ),
        key=lambda row: (row.sort_order, row.category),
    )
    return rows, datetime.now(timezone.utc)

resolve_site_any(site)

Resolve a site by numeric id first, then acronym/domain. Used by v2.

Source code in src/cms_api/public/store/memory.py

def resolve_site_any(self, site: str) -> SiteOut:
    """Resolve a site by numeric id first, then acronym/domain. Used by v2."""
    try:
        site_id = int(site)
    except (TypeError, ValueError):
        site_id = None
    if site_id is not None and site_id in self.sites:
        row = self.sites[site_id]
        if not row.is_active:
            raise APIError("resource_not_found", "Site not found", 404, {"site": site})
        return row
    for row in self.sites.values():
        if row.acronym == site or row.domain == site:
            if not row.is_active:
                raise APIError("resource_not_found", "Site not found", 404, {"site": site})
            return row
    raise APIError("resource_not_found", "Site not found", 404, {"site": site})

get_template_css(slug)

Return (css, css_version) for a template by slug (for the CSS endpoint).

Source code in src/cms_api/public/store/memory.py

def get_template_css(self, slug: str) -> Tuple[str, Optional[str]]:
    """Return ``(css, css_version)`` for a template by slug (for the CSS endpoint)."""
    for template in self.templates.values():
        if getattr(template, "slug", None) == slug:
            return (getattr(template, "css", None) or "", getattr(template, "css_version", None))
    raise APIError("resource_not_found", "Template not found", 404, {"slug": slug})

build_settings_blob(site, language=None)

Build the v2 settings bundle (identity + merged settings + templates).

Source code in src/cms_api/public/store/memory.py

def build_settings_blob(
    self, site: str, language: Optional[str] = None
) -> Tuple[SettingsBundle, Optional[datetime]]:
    """Build the v2 settings bundle (identity + merged settings + templates)."""
    resolved_site = self.resolve_site_any(site)
    settings = self._effective_settings_v2(resolved_site.id)
    bundle = v2_common.build_settings_blob(
        site=resolved_site,
        settings=settings,
        categories=self._categories_light_v2(resolved_site.id),
        tags=self._tags_light_v2(resolved_site.id),
        templates=self._resolved_templates_map_v2(resolved_site),
        active_template=self._active_template_v2(resolved_site),
        feature_flags={},
        last_modified=None,
        page_types=self._resolved_page_types_v2(resolved_site),
        authors=self._authors_light_v2(resolved_site.id),
    )
    return bundle, None

get_article_page(site_id, slug, language=None, version=None, related_limit=6)

Build the v2 article page payload, optionally from content_history.

Source code in src/cms_api/public/store/memory.py

def get_article_page(
    self,
    site_id: str,
    slug: str,
    language: Optional[str] = None,
    version: Optional[int] = None,
    related_limit: int = 6,
) -> Tuple[ArticlePage, Optional[datetime]]:
    """Build the v2 article page payload, optionally from content_history."""
    site = self.resolve_site_any(site_id)
    live = self._published_article(site.id, slug, language)
    settings = self._effective_settings_v2(site.id)
    cdn = settings.get("cdn_base_url")
    snapshot_changed_at: Optional[datetime] = None
    if version:
        row, snapshot_changed_at = self._merged_history(live, version)
    else:
        row = live
    author = self.authors.get(row.author_id)
    if author is None or not author.is_active:
        raise APIError("resource_not_found", "Author not found", 404, {"author_id": row.author_id})
    content = v2_common.build_article_content(row, row.categories, row.tags, cdn)
    author_slice = v2_common.author_slice(author, cdn)
    seo = v2_common.build_article_seo(row, author.name, cdn)
    breadcrumbs = v2_common.build_breadcrumbs(
        site.name, row.categories[0] if row.categories else None, content.title
    )
    related = self._related_v2(site.id, live, related_limit, cdn)
    # On a versioned read the snapshot's change timestamp is the resource's
    # last-modified; fall back to the live row only when it is null.
    if version:
        last_mod = snapshot_changed_at or content.updated_at or content.published_at
    else:
        last_mod = content.updated_at or content.published_at
    page = ArticlePage(
        content=content,
        author=author_slice,
        seo=seo,
        breadcrumbs=breadcrumbs,
        related=related,
        meta=v2_common.meta_envelope("pages", last_mod),
    )
    return page, last_mod

get_author_profile(site_id, slug)

Build the v2 author profile payload with article stats.

Source code in src/cms_api/public/store/memory.py

def get_author_profile(self, site_id: str, slug: str) -> Tuple[AuthorProfile, Optional[datetime]]:
    """Build the v2 author profile payload with article stats."""
    site = self.resolve_site_any(site_id)
    author = None
    for candidate in self.authors.values():
        if candidate.site_id == site.id and candidate.slug == slug and candidate.is_active:
            author = candidate
            break
    if author is None:
        raise APIError("resource_not_found", "Author not found", 404, {"slug": slug})
    settings = self._effective_settings_v2(site.id)
    cdn = settings.get("cdn_base_url")
    published = [
        item
        for item in self.content.values()
        if item.site_id == site.id and item.author_id == author.id and item.status == _PUBLISHED
    ]
    count = len(published)
    latest = max((item.published_at for item in published if item.published_at), default=None)
    author_slice = v2_common.author_slice(author, cdn)
    seo = v2_common.build_author_seo(author_slice, f"/author/{author.slug}")
    stats = v2_common.author_stats(count, latest)
    profile = AuthorProfile(
        author=author_slice, seo=seo, stats=stats, meta=v2_common.meta_envelope("authors", latest)
    )
    return profile, latest

list_items(site_id, kind, params, language=None, limit=20, cursor=None)

Return metadata-only content items for a list kind, cursor-paginated.

Source code in src/cms_api/public/store/memory.py

def list_items(
    self,
    site_id: str,
    kind: str,
    params: Dict[str, Any],
    language: Optional[str] = None,
    limit: int = 20,
    cursor: Optional[str] = None,
) -> Tuple[ListResponse, Optional[datetime]]:
    """Return metadata-only content items for a list ``kind``, cursor-paginated."""
    site = self.resolve_site_any(site_id)
    settings = self._effective_settings_v2(site.id)
    cdn = settings.get("cdn_base_url")
    matched = [
        item
        for item in self.content.values()
        if item.site_id == site.id
        and item.status == _PUBLISHED
        and item.is_active
        and (language is None or item.language == language)
        and self._matches_kind(item, kind, params, site.id)
    ]
    matched.sort(
        key=lambda item: (item.published_at or datetime.min.replace(tzinfo=timezone.utc), item.id),
        reverse=True,
    )
    ids = [item.id for item in matched]
    page_ids, next_cursor = paginate_ids(ids, limit, cursor)
    by_id = {item.id: item for item in matched}
    items = []
    for item_id in page_ids:
        item = by_id[item_id]
        author = self.authors.get(item.author_id)
        items.append(
            v2_common.list_item_from_fields(
                id=item.id,
                slug=item.slug,
                title=item.title,
                description=item.description,
                featured_image=item.featured_image,
                **v2_common.featured_image_variants(item),
                author_name=author.name if author else None,
                author_image=author.image if author else None,
                category_slug=item.categories[0] if item.categories else None,
                published_at=item.published_at,
                cdn_base_url=cdn,
            )
        )
    last_mod = max((by_id[i].published_at for i in page_ids if by_id[i].published_at), default=None)
    response = ListResponse(
        items=items, next_cursor=next_cursor, total=None, meta=v2_common.meta_envelope("lists", last_mod)
    )
    return response, last_mod

build_sitemap(site_id, since=None, limit=5000, cursor=None)

Build the v2 sitemap (flagged content + static pages), cursor-paginated.

Source code in src/cms_api/public/store/memory.py

def build_sitemap(
    self,
    site_id: str,
    since: Optional[str] = None,
    limit: int = 5000,
    cursor: Optional[str] = None,
) -> Tuple[SitemapResponse, Optional[datetime]]:
    """Build the v2 sitemap (flagged content + static pages), cursor-paginated."""
    site = self.resolve_site_any(site_id)
    settings = self._effective_settings_v2(site.id)
    since_dt = v2_common.parse_since(since)
    flagged = [
        item
        for item in self.content.values()
        if item.site_id == site.id
        and item.status == _PUBLISHED
        and item.is_active
        and (item.is_featured or item.show_in_homepage)
        and (since_dt is None or (item.published_at and item.published_at >= since_dt))
    ]
    flagged.sort(
        key=lambda item: (item.published_at or datetime.min.replace(tzinfo=timezone.utc), item.id),
        reverse=True,
    )
    ids = [item.id for item in flagged]
    page_ids, next_cursor = paginate_ids(ids, limit, cursor)
    by_id = {item.id: item for item in flagged}
    entries = [
        v2_common.sitemap_entry(
            loc=v2_common.compose_loc(by_id[i].categories[0] if by_id[i].categories else None, by_id[i].slug),
            lastmod=by_id[i].published_at,
            changefreq="daily",
            priority=0.8,
        )
        for i in page_ids
    ]
    static_entries = []
    if not cursor:
        nav_urls = v2_common.navigation_urls(settings.get("navigation") or [])
        page_names = {name for name, el in self.template_elements.items() if el.get("type") == "page"}
        static_entries = v2_common.static_sitemap_entries(page_names, nav_urls)
    all_entries = static_entries + entries
    last_mod = max((by_id[i].published_at for i in page_ids if by_id[i].published_at), default=None)
    response = SitemapResponse(
        items=all_entries, next_cursor=next_cursor, total=None, meta=v2_common.meta_envelope("sitemap", last_mod)
    )
    return response, last_mod

Shared shapers (v2_common)

v2_common

Shared payload shapers for the public v2 surface (AISD-116).

Both the SQL and in-memory stores fetch raw rows their own way, then call these pure helpers to build the v2 response envelopes. Keeping the shaping logic here guarantees the two stores return byte-identical shapes, which is what lets the memory-backed tests validate the SQL store's contract.

cache_hint(endpoint)

Return the CacheHint model for an endpoint name.

Source code in src/cms_api/public/store/v2_common.py

def cache_hint(endpoint: str) -> CacheHint:
    """Return the CacheHint model for an endpoint name."""
    ttl, swr = CACHE_HINTS[endpoint]
    return CacheHint(ttl_seconds=ttl, swr_seconds=swr)

meta_envelope(endpoint, last_modified)

Build the MetaEnvelope for a response (etag is filled by the cache layer).

Source code in src/cms_api/public/store/v2_common.py

def meta_envelope(endpoint: str, last_modified: Optional[datetime]) -> MetaEnvelope:
    """Build the MetaEnvelope for a response (etag is filled by the cache layer)."""
    return MetaEnvelope(etag=None, last_modified=last_modified, cache_hint=cache_hint(endpoint))

parse_since(since)

Parse an RFC1123 / ISO since query value to an aware UTC datetime.

Source code in src/cms_api/public/store/v2_common.py

def parse_since(since: Optional[str]) -> Optional[datetime]:
    """Parse an RFC1123 / ISO ``since`` query value to an aware UTC datetime."""
    if not since:
        return None
    parsed: Optional[datetime]
    try:
        parsed = parsedate_to_datetime(since)
    except (TypeError, ValueError):
        parsed = None
    if parsed is None:
        try:
            parsed = datetime.fromisoformat(since.replace("Z", "+00:00"))
        except ValueError:
            return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed

coerce_page_types(raw)

Validate raw page-type descriptor dicts into PageTypeDescriptor models.

A malformed descriptor (bad call, missing routes, unknown field, ...) is dropped with a logged warning rather than failing the whole /settings response -- one bad row must never brick every page on the site. This is the server-side validation gate for descriptors authored in the DB / admin.

Source code in src/cms_api/public/store/v2_common.py

def coerce_page_types(raw: Sequence[Dict[str, Any]]) -> List[PageTypeDescriptor]:
    """Validate raw page-type descriptor dicts into PageTypeDescriptor models.

    A malformed descriptor (bad ``call``, missing ``routes``, unknown field, ...)
    is dropped with a logged warning rather than failing the whole ``/settings``
    response -- one bad row must never brick every page on the site. This is the
    server-side validation gate for descriptors authored in the DB / admin.
    """
    out: List[PageTypeDescriptor] = []
    for entry in raw:
        try:
            out.append(PageTypeDescriptor.model_validate(entry))
        except ValidationError as exc:
            logger.warning(
                "skipping invalid page_type descriptor type=%s variant=%s: %s",
                (entry or {}).get("type"),
                (entry or {}).get("variant"),
                exc,
            )
    # Stable order for the worker matcher: ascending position (None last), then type.
    out.sort(key=lambda d: (d.position is None, d.position or 0, d.type, d.variant))
    return out

build_settings_blob(site, settings, categories, tags, templates, active_template, feature_flags, last_modified, page_types=(), authors=())

Assemble the SettingsBundle from raw row tuples (data-access agnostic).

active_template is (id, slug, version, css_version); page_types is a sequence of raw descriptor dicts validated via :func:coerce_page_types. authors is a sequence of dicts (id, slug, name, initials, image, bio, article_count) for the site-wide authors index; image is absolutized here.

Source code in src/cms_api/public/store/v2_common.py

def build_settings_blob(
    site: Any,
    settings: Dict[str, Any],
    categories: Sequence[Tuple[int, str, str, int]],
    tags: Sequence[Tuple[int, str, str]],
    templates: Dict[str, List[Tuple[Optional[str], Optional[Any]]]],
    active_template: Optional[Tuple[int, str, Optional[str], Optional[str]]],
    feature_flags: Optional[Dict[str, Any]],
    last_modified: Optional[datetime],
    page_types: Sequence[Dict[str, Any]] = (),
    authors: Sequence[Dict[str, Any]] = (),
) -> SettingsBundle:
    """Assemble the SettingsBundle from raw row tuples (data-access agnostic).

    ``active_template`` is ``(id, slug, version, css_version)``; ``page_types`` is
    a sequence of raw descriptor dicts validated via :func:`coerce_page_types`.
    ``authors`` is a sequence of dicts (id, slug, name, initials, image, bio,
    article_count) for the site-wide authors index; image is absolutized here.
    """
    cdn = settings.get("cdn_base_url")
    blob_kwargs: Dict[str, Any] = {}
    for field in SETTINGS_FIELDS:
        value = settings.get(field)
        if field in _MEDIA_SETTINGS_FIELDS:
            value = absolutize_media_url(value, cdn)
        blob_kwargs[field] = value
    blob_kwargs["maintenance_mode"] = bool(settings.get("maintenance_mode"))

    blob_kwargs["categories"] = [
        CategoryLight(id=cid, slug=slug, name=name, sort_order=sort_order or 0)
        for (cid, slug, name, sort_order) in categories
    ]
    blob_kwargs["tags"] = [TagLight(id=tid, slug=slug, name=name) for (tid, slug, name) in tags]
    blob_kwargs["authors"] = [
        AuthorLight(
            id=a["id"],
            slug=a["slug"],
            name=a["name"],
            initials=a.get("initials"),
            image=absolutize_media_url(a.get("image"), cdn),
            bio=a.get("bio"),
            article_count=int(a.get("article_count") or 0),
        )
        for a in authors
    ]
    blob_kwargs["templates"] = {
        element_type: [
            TemplateBlockResolved(html_content=html, settings=block_settings)
            for (html, block_settings) in blocks
        ]
        for element_type, blocks in templates.items()
    }
    blob_kwargs["active_template"] = (
        ActiveTemplate(
            id=active_template[0],
            slug=active_template[1],
            version=active_template[2],
            css_version=active_template[3] if len(active_template) > 3 else None,
        )
        if active_template
        else None
    )
    blob_kwargs["page_types"] = coerce_page_types(page_types)
    blob_kwargs["feature_flags"] = feature_flags or {}

    identity = SiteIdentity(
        id=site.id,
        acronym=site.acronym,
        name=site.name,
        domain=site.domain,
        default_language=site.default_language,
        tagline=site.tagline,
    )
    return SettingsBundle(
        site=identity,
        site_settings=SiteSettingsBlob(**blob_kwargs),
        meta=meta_envelope("settings", last_modified),
    )

featured_image_variants(row)

Return article-level image paths from featured_image + settings.

Source code in src/cms_api/public/store/v2_common.py

def featured_image_variants(row: Any) -> Dict[str, Optional[str]]:
    """Return article-level image paths from ``featured_image`` + ``settings``."""
    settings = row.settings if isinstance(getattr(row, "settings", None), dict) else {}
    image = getattr(row, "featured_image", None)
    return {
        "image": image,
        "image_opt": settings.get("featured_image_opt") or image,
        "image_sm": settings.get("featured_image_sm") or image,
    }

build_article_content(row, categories, tags, cdn_base_url, updated_at=None)

Build the ArticleContent from a content row plus its categories/tags.

Works for both the ORM Content row and the admin ContentOut model since their flat columns share names.

Source code in src/cms_api/public/store/v2_common.py

def build_article_content(
    row: Any,
    categories: Sequence[str],
    tags: Sequence[str],
    cdn_base_url: Optional[str],
    updated_at: Optional[datetime] = None,
) -> ArticleContent:
    """Build the ArticleContent from a content row plus its categories/tags.

    Works for both the ORM ``Content`` row and the admin ``ContentOut`` model
    since their flat columns share names.
    """
    variants = featured_image_variants(row)
    return ArticleContent(
        id=row.id,
        site_id=row.site_id,
        content_type=row.content_type,
        language=row.language,
        slug=row.slug,
        title=row.title,
        description=row.description,
        author_id=row.author_id,
        featured_image=absolutize_media_url(row.featured_image, cdn_base_url),
        image=absolutize_media_url(variants["image"], cdn_base_url),
        image_opt=absolutize_media_url(variants["image_opt"], cdn_base_url),
        image_sm=absolutize_media_url(variants["image_sm"], cdn_base_url),
        show_in_homepage=bool(row.show_in_homepage),
        is_featured=bool(row.is_featured),
        is_trending=bool(row.is_trending),
        content_blocks=row.content_blocks,
        settings=row.settings,
        published_at=row.published_at,
        updated_at=updated_at if updated_at is not None else getattr(row, "updated_at", None),
        categories=list(categories),
        tags=list(tags),
    )

build_article_seo(row, author_name, cdn_base_url)

Build the SEO block for an article row, computing structured_data.

Source code in src/cms_api/public/store/v2_common.py

def build_article_seo(row: Any, author_name: Optional[str], cdn_base_url: Optional[str]) -> SeoBlock:
    """Build the SEO block for an article row, computing structured_data."""
    og_image = getattr(row, "og_image", None) or getattr(row, "featured_image", None)
    seo = SeoBlock(
        seo_title=getattr(row, "seo_title", None) or row.title,
        seo_description=getattr(row, "seo_description", None) or getattr(row, "description", None),
        og_title=getattr(row, "og_title", None),
        og_description=getattr(row, "og_description", None),
        og_image=absolutize_media_url(og_image, cdn_base_url),
        canonical_url=getattr(row, "canonical_url", None),
        noindex=bool(getattr(row, "noindex", False)),
        schema_type=getattr(row, "schema_type", None),
    )
    seo.structured_data = _structured_data_article(row, author_name)
    return seo

build_author_seo(author, canonical_url)

Build the computed SEO block for an author profile.

Source code in src/cms_api/public/store/v2_common.py

def build_author_seo(author: AuthorSlice, canonical_url: Optional[str]) -> SeoBlock:
    """Build the computed SEO block for an author profile."""
    structured = {
        "@context": "https://schema.org",
        "@type": "Person",
        "name": author.name,
    }
    if author.image:
        structured["image"] = author.image
    return SeoBlock(
        seo_title=author.name,
        seo_description=author.bio,
        og_image=author.image,
        canonical_url=canonical_url,
        noindex=False,
        schema_type="Person",
        structured_data=structured,
    )

build_breadcrumbs(site_name, primary_category, title)

Compose breadcrumbs from site name, first category, and the article title.

Source code in src/cms_api/public/store/v2_common.py

def build_breadcrumbs(site_name: str, primary_category: Optional[str], title: str) -> List[NavItem]:
    """Compose breadcrumbs from site name, first category, and the article title."""
    crumbs = [NavItem(label="Home", url="/")]
    if primary_category:
        crumbs.append(NavItem(label=primary_category, url=f"/{primary_category}"))
        crumbs.append(NavItem(label=title, url=f"/{primary_category}/{_slug_or_blank(title)}"))
    else:
        crumbs.append(NavItem(label=title, url=f"/{_slug_or_blank(title)}"))
    return crumbs

list_item_from_fields(*, id, slug, title, description, featured_image, image=None, image_opt=None, image_sm=None, author_name, author_image, category_slug, published_at, cdn_base_url)

Build a metadata-only ListItem, absolutizing media URLs.

Source code in src/cms_api/public/store/v2_common.py

def list_item_from_fields(
    *,
    id: int,
    slug: str,
    title: str,
    description: Optional[str],
    featured_image: Optional[str],
    image: Optional[str] = None,
    image_opt: Optional[str] = None,
    image_sm: Optional[str] = None,
    author_name: Optional[str],
    author_image: Optional[str],
    category_slug: Optional[str],
    published_at: Optional[datetime],
    cdn_base_url: Optional[str],
) -> ListItem:
    """Build a metadata-only ListItem, absolutizing media URLs."""
    origin = image or featured_image
    return ListItem(
        id=id,
        slug=slug,
        title=title,
        description=description,
        featured_image=absolutize_media_url(featured_image, cdn_base_url),
        image=absolutize_media_url(origin, cdn_base_url),
        image_opt=absolutize_media_url(image_opt or origin, cdn_base_url),
        image_sm=absolutize_media_url(image_sm or origin, cdn_base_url),
        author_name=author_name,
        author_image=absolutize_media_url(author_image, cdn_base_url),
        category_slug=category_slug,
        published_at=published_at,
    )

author_slice(author, cdn_base_url)

Build the AuthorSlice from an author row, absolutizing the image URL.

Source code in src/cms_api/public/store/v2_common.py

def author_slice(author: Any, cdn_base_url: Optional[str]) -> AuthorSlice:
    """Build the AuthorSlice from an author row, absolutizing the image URL."""
    return AuthorSlice(
        id=author.id,
        slug=author.slug,
        name=author.name,
        initials=author.initials,
        image=absolutize_media_url(author.image, cdn_base_url),
        bio=author.bio,
    )

author_stats(article_count, latest_published_at)

Build the AuthorStats counters.

Source code in src/cms_api/public/store/v2_common.py

def author_stats(article_count: int, latest_published_at: Optional[datetime]) -> AuthorStats:
    """Build the AuthorStats counters."""
    return AuthorStats(article_count=article_count, latest_published_at=latest_published_at)

related_item(*, id, slug, title, featured_image, author_name, published_at, cdn_base_url)

Build a RelatedItem, absolutizing the featured image.

Source code in src/cms_api/public/store/v2_common.py

def related_item(
    *,
    id: int,
    slug: str,
    title: str,
    featured_image: Optional[str],
    author_name: Optional[str],
    published_at: Optional[datetime],
    cdn_base_url: Optional[str],
) -> RelatedItem:
    """Build a RelatedItem, absolutizing the featured image."""
    return RelatedItem(
        id=id,
        slug=slug,
        title=title,
        featured_image=absolutize_media_url(featured_image, cdn_base_url),
        author_name=author_name,
        published_at=published_at,
    )

sitemap_entry(*, loc, lastmod, changefreq, priority)

Build a SitemapEntry.

Source code in src/cms_api/public/store/v2_common.py

def sitemap_entry(
    *, loc: str, lastmod: Optional[datetime], changefreq: str, priority: float
) -> SitemapEntry:
    """Build a SitemapEntry."""
    return SitemapEntry(loc=loc, lastmod=lastmod, changefreq=changefreq, priority=priority)

settings_last_modified(*updated_ats)

Return the latest of several updated_at timestamps (for settings meta).

Source code in src/cms_api/public/store/v2_common.py

def settings_last_modified(*updated_ats: Optional[datetime]) -> Optional[datetime]:
    """Return the latest of several updated_at timestamps (for settings meta)."""
    return _max_datetime(updated_ats)

compose_loc(category, slug)

Compose a content URL path from its primary category and slug.

Source code in src/cms_api/public/store/v2_common.py

def compose_loc(category: Optional[str], slug: str) -> str:
    """Compose a content URL path from its primary category and slug."""
    if category:
        return f"/{category}/{slug}"
    return f"/{slug}"

navigation_urls(navigation)

Flatten navigation entries (with children) to a list of URL strings.

Source code in src/cms_api/public/store/v2_common.py

def navigation_urls(navigation: Any) -> List[str]:
    """Flatten navigation entries (with children) to a list of URL strings."""
    urls: List[str] = []
    if not isinstance(navigation, list):
        return urls
    for item in navigation:
        if not isinstance(item, dict):
            continue
        url = item.get("url") or item.get("href")
        if url:
            urls.append(url)
        urls.extend(navigation_urls(item.get("children")))
    return urls

url_to_element_name(url)

Map a nav URL path to its template_element name (/contact-us -> contact_us).

Source code in src/cms_api/public/store/v2_common.py

def url_to_element_name(url: str) -> str:
    """Map a nav URL path to its template_element name (``/contact-us`` -> ``contact_us``)."""
    return url.strip("/").replace("-", "_")

static_sitemap_entries(page_element_names, nav_urls)

Build sitemap entries for nav URLs backed by a type='page' template element.

The site homepage (/) is always included; other nav URLs are included only when they correspond to a known static-page template element.

Source code in src/cms_api/public/store/v2_common.py

def static_sitemap_entries(page_element_names: set[str], nav_urls: List[str]) -> list:
    """Build sitemap entries for nav URLs backed by a ``type='page'`` template element.

    The site homepage (``/``) is always included; other nav URLs are included only
    when they correspond to a known static-page template element.
    """
    entries = []
    seen: set[str] = set()
    for url in nav_urls:
        if url in seen:
            continue
        is_home = url == "/"
        if not is_home and url_to_element_name(url) not in page_element_names:
            continue
        seen.add(url)
        entries.append(
            sitemap_entry(
                loc=url,
                lastmod=None,
                changefreq="hourly" if is_home else "weekly",
                priority=1.0 if is_home else 0.5,
            )
        )
    return entries