core

core

Core DomoPage entity class and basic operations.

DomoPage dataclass

DomoPage(
    auth: DomoAuth,
    id: int,
    raw: dict,
    Lineage: DomoLineage | None = None,
    title: str | None = None,
    top_page_id: int | None = None,
    parent_page_id: int | None = None,
    is_locked: bool | None = None,
    collections: list[dict] = list(),
    custom_attributes: dict[str, Any] = dict(),
    parent_page: dict[str, Any] | None = None,
    top_page: dict[str, Any] | None = None,
    children: list[Any] = list(),
    Access: DomoPageAccessController | None = None,
    Layout: DomoPageLayout | None = None,
)

Bases: DomoPage_Default

DomoPage entity with federation support via FederationContext.

create async classmethod

create(
    auth: DomoAuth,
    title: str,
    parent_page_id: int = 0,
    *,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs
)

Create a new page (dashboard) and return the hydrated DomoPage.

For a full dashboard (page + cards + layout) use dashboard_builder.create_dashboard; this creates the page shell.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	DomoAuth instance.	required
title	str	Page title.	required
parent_page_id	int	Parent page id (0 for a top-level page).	0
return_raw	bool	Return the raw create response instead of a DomoPage.	False
context	RouteContext | None	Optional RouteContext.	None

Source code in src/crew_dcs/classes/DomoPage/core.py

@classmethod
async def create(
    cls,
    auth: DomoAuth,
    title: str,
    parent_page_id: int = 0,
    *,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Create a new page (dashboard) and return the hydrated DomoPage.

    For a full dashboard (page + cards + layout) use
    ``dashboard_builder.create_dashboard``; this creates the page shell.

    Args:
        auth: DomoAuth instance.
        title: Page title.
        parent_page_id: Parent page id (0 for a top-level page).
        return_raw: Return the raw create response instead of a DomoPage.
        context: Optional RouteContext.
    """
    # create + hydrate reuse one pooled session via auth-level connection
    # reuse (RouteContext.reuse_session, default on).
    res = await page_routes.create_page(
        auth=auth,
        title=title,
        parent_page_id=parent_page_id,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    page_id = res.response.get("pageId") if isinstance(res.response, dict) else None
    if not page_id:
        raise ValueError(
            f"create_page response did not contain a pageId: {res.response!r}"
        )

    return await cls.get_by_id(page_id=str(page_id), auth=auth, context=context)

delete async

delete(
    *, context: RouteContext | None = None, **context_kwargs
)

Delete this page.

Source code in src/crew_dcs/classes/DomoPage/core.py

async def delete(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Delete this page."""
    context = RouteContext.build_context(context=context, **context_kwargs)
    return await page_routes.delete_page(
        auth=self.auth, page_id=str(self.id), context=context
    )

get_by_id async classmethod

get_by_id(
    page_id: str,
    auth: DomoAuth,
    suppress_no_results_error: bool = True,
    return_raw: bool = False,
    debug_api: bool = False,
    include_layout: bool = False,
    debug_num_stacks_to_drop=2,
    session: AsyncClient = None,
    id=None,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    max_subscriptions_to_check: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Retrieve page data as a DomoPage instance.

Parameters:

Name	Type	Description	Default
page_id	str	Page identifier	required
auth	DomoAuth	Authentication object	required
suppress_no_results_error	bool	Suppress errors when no results found	True
return_raw	bool	Return raw API response	False
debug_api	bool	Enable debug logging	False
include_layout	bool	Include page layout in response	False
debug_num_stacks_to_drop		Stack frames to drop in debug output	2
session	AsyncClient	Optional HTTP session	None
id		Alternative parameter for page_id	None
check_if_published	bool | None	When True, checks if page is published (requires subscription check). If None and either parent_auth or parent_auth_retrieval_fn is provided, defaults to True.	None
parent_auth_retrieval_fn	Callable | None	Callable returning publisher auth when given publisher domain	None
parent_auth	DomoAuth | None	Pre-existing publisher auth (alternative to parent_auth_retrieval_fn)	None
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
	DomoPage instance

Source code in src/crew_dcs/classes/DomoPage/core.py

@classmethod
async def get_by_id(
    cls,
    page_id: str,
    auth: DomoAuth,
    suppress_no_results_error: bool = True,
    return_raw: bool = False,
    debug_api: bool = False,
    include_layout: bool = False,
    debug_num_stacks_to_drop=2,
    session: httpx.AsyncClient = None,
    id=None,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    max_subscriptions_to_check: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Retrieve page data as a DomoPage instance.

    Args:
        page_id: Page identifier
        auth: Authentication object
        suppress_no_results_error: Suppress errors when no results found
        return_raw: Return raw API response
        debug_api: Enable debug logging
        include_layout: Include page layout in response
        debug_num_stacks_to_drop: Stack frames to drop in debug output
        session: Optional HTTP session
        id: Alternative parameter for page_id
        check_if_published: When True, checks if page is published (requires subscription check).
            If None and either parent_auth or parent_auth_retrieval_fn is provided, defaults to True.
        parent_auth_retrieval_fn: Callable returning publisher auth when given publisher domain
        parent_auth: Pre-existing publisher auth (alternative to parent_auth_retrieval_fn)
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        DomoPage instance
    """
    # Auto-enable publish checking when parent auth is provided (either directly or via retrieval fn)
    # unless explicitly disabled
    if (parent_auth_retrieval_fn or parent_auth) and check_if_published is None:
        check_if_published = True

    page_id = page_id or id

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        parent_class=cls.__name__,
        **context_kwargs,
    )

    # Get page data
    res = await page_routes.get_page_by_id(
        auth=auth,
        page_id=page_id,
        include_layout=include_layout,
        context=context,
    )

    if return_raw:
        return res

    is_published = False
    subscription = None

    if check_if_published and (parent_auth_retrieval_fn or parent_auth):
        # Create a retrieval function from parent_auth if only parent_auth was provided
        effective_retrieval_fn = parent_auth_retrieval_fn
        if not effective_retrieval_fn and parent_auth:
            effective_retrieval_fn = lambda _domain: parent_auth  # noqa: E731

        # Local import to avoid circular dependency
        from ..subentity.lineage.federation_context import FederationContext

        probe = FederationContext.from_entity_id(
            auth=auth,
            entity_id=str(page_id),
            entity_type="PAGE",
        )
        is_published = await probe.check_if_published(
            retrieve_parent_auth_fn=effective_retrieval_fn,
            entity_type="PAGE",
            session=session,
            debug_api=debug_api,
            max_subscriptions_to_check=max_subscriptions_to_check,
        )
        if is_published:
            subscription = probe.subscription

    # Fetch owners if present in response
    owners = None
    dd = (
        util_dd.DictDot(res.response)
        if isinstance(res.response, dict)
        else res.response
    )
    if hasattr(dd, "page") and dd.page.owners and len(dd.page.owners) > 0:
        pg_temp = DomoPage_Default(
            id=int(dd.id),
            title=dd.title,
            raw=res.response,
            parent_page_id=(
                int(dd.page.parentPageId) if dd.page.parentPageId else None
            ),
            collections=dd.collections,
            auth=auth,
        )
        owners = await pg_temp._get_domo_owners_from_dd(
            dd.page.owners,
            suppress_no_results_error=suppress_no_results_error,
            debug_api=debug_api,
            session=session,
            debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        )

    # Create page instance
    pg = cls._from_content_stacks_v3(
        page_obj=res.response,
        auth=auth,
        owners=owners,
    )

    if is_published and subscription:
        helper = pg.enable_federation_support()
        helper.hydrate_from_existing(
            subscription=subscription,
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            parent_auth=parent_auth,
            content_type="PAGE",
            entity_id=str(page_id),
        )

    # Auto-trace lineage if parent_auth_retrieval_fn is provided
    if parent_auth_retrieval_fn and pg.Lineage:
        await pg.Lineage.get(
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            parent_auth=parent_auth,
            session=session,
            debug_api=debug_api,
            context=context,
        )

    pg.custom_attributes["parent_page"] = None
    pg.custom_attributes["top_page"] = None

    return pg

DomoPage_Default dataclass

DomoPage_Default(
    auth: DomoAuth,
    id: int,
    raw: dict,
    Lineage: DomoLineage | None = None,
    title: str | None = None,
    top_page_id: int | None = None,
    parent_page_id: int | None = None,
    is_locked: bool | None = None,
    collections: list[dict] = list(),
    custom_attributes: dict[str, Any] = dict(),
    parent_page: dict[str, Any] | None = None,
    top_page: dict[str, Any] | None = None,
    children: list[Any] = list(),
    Access: DomoPageAccessController | None = None,
    Layout: DomoPageLayout | None = None,
)

Bases: DomoEntity_w_Lineage

entity_name property

entity_name: str

Get the display name for this page.

Pages use the 'title' field as their display name.

Returns:

Type	Description
str	Page title, or page ID as fallback

is_federated property

is_federated: bool

Check if this page is federated (published/subscribed).

name property

name: str

Get the display name for this page.

Implements abstract property from DomoEntity_w_Lineage. Pages use the 'title' field as their display name.

Returns:

Type	Description
str	Page title, or "Untitled Page {id}" as fallback

to_dict

to_dict(
    override_fn: Callable | None = None,
    return_snake_case: bool = False,
) -> dict

Serialize page for export/debugging with layout card details.

Source code in src/crew_dcs/classes/DomoPage/core.py

def to_dict(
    self, override_fn: Callable | None = None, return_snake_case: bool = False
) -> dict:
    """Serialize page for export/debugging with layout card details."""
    if override_fn:
        return override_fn(self)

    result = super().to_dict(return_snake_case=return_snake_case)

    layout_key = "layout" if return_snake_case else "Layout"
    cards_key = "cards" if return_snake_case else "Cards"

    def _serialize_card(card: Any) -> Any:
        return self._serialize_related_entity(
            card,
            return_snake_case=return_snake_case,
            include_entity_type=True,
        )

    if self.Layout and self.Layout.cards:
        layout_payload: dict[str, Any] = {
            cards_key: [
                c
                for c in (_serialize_card(c) for c in self.Layout.cards)
                if c is not None
            ]
        }
        result[layout_key] = layout_payload

    return result