DomoPage

DomoPage ¶

Page classes and functionality.

This module provides comprehensive page management functionality for Domo pages, organized using the subentity composition pattern:

exceptions: Page-related exception classes
core: Main DomoPage entity classes (DomoPage_Default, DomoPage factory)
pages: DomoPages collection and hierarchy operations
layout: Layout and content management subentity (DomoPageLayout)

Classes:

Name	Description
`DomoPage`	Factory class for page operations
`DomoPage_Default`	Default page implementation
`DomoPages`	Collection class for managing multiple pages
`DomoPageAccessController`	Access control subentity
`DomoPageLayout`	Layout and content management subentity

Example

Basic page usage with subentities:

>>> page = await DomoPage.get_by_id(page_id="123", auth=auth)
>>> access_list = await page.Access.get_accesslist()
>>> owners = await page.Access.get_owners()

>>> # Get cards
>>> cards = await page.Layout.get_cards()
>>> # Update layout
>>> await page.Layout.update(body={...})

DomoLineageLink_Page `dataclass` ¶

DomoLineageLink_Page(
    auth: DomoAuth,
    id: str,
    entity: Any,
    _type: str | None = None,
    dependencies: list[DomoLineage_Link] = list(),
    dependents: list[DomoLineage_Link] = list(),
)

Bases: DomoLineage_Link

get_entity `async` ¶

get_entity(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context=None
)

Get the entity associated with this lineage link.

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

async def get_entity(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context=None,
):
    """Get the entity associated with this lineage link."""
    from .core import DomoPage

    if context is None:
        context = RouteContext(
            session=session,
            debug_api=debug_api,
        )

    return await DomoPage.get_by_id(
        page_id=self.id,
        auth=self.auth,
        context=context,
    )

DomoLineage_Page `dataclass` ¶

DomoLineage_Page(
    auth: DomoAuth,
    parent: Any = None,
    lineage: list[DomoLineage_Link] = list(),
    immediate_dependencies: list[DomoLineage_Link] = list(),
    immediate_dependents: list[DomoLineage_Link] = list(),
    downstream_lineage: list[DomoLineage_Link] = list(),
    cards: list[Any] = None,
)

Bases: DomoLineage

Lineage handler for page entities.

The parent is the page entity this lineage is based off of (not a dependency). Pages have cards as their immediate dependencies.

get `async` ¶

get(
    session: AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    parent_auth: DomoAuth = None,
    parent_auth_retrieval_fn: (
        Callable[[str], Any] | None
    ) = None,
    max_depth: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoLineage_Link]

Get lineage for a page by traversing card lineage.

Pages do not have direct datacenter lineage. Instead, we load cards from the page layout and then fetch each card's lineage. Each card is added to the page lineage as a LineageLink with dependencies and dependents referenced from the card's lineage root node.

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

async def get(
    self,
    session: httpx.AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    parent_auth: DomoAuth = None,
    parent_auth_retrieval_fn: Callable[[str], Any] | None = None,
    max_depth: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoLineage_Link]:
    """Get lineage for a page by traversing card lineage.

    Pages do not have direct datacenter lineage. Instead, we load cards
    from the page layout and then fetch each card's lineage. Each card
    is added to the page lineage as a LineageLink with dependencies and
    dependents referenced from the card's lineage root node.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    if not self.parent:
        raise ClassError(
            message="Parent must be set. Use from_parent() to create lineage with a parent.",
            cls_instance=self,
        )

    cache_key = self._build_cache_key(
        parent_auth=parent_auth,
        parent_auth_retrieval_fn=parent_auth_retrieval_fn,
        return_raw=return_raw,
    )

    if self.lineage and self._cached_lineage_params == cache_key:
        await logger.debug(
            f"DomoLineage_Page.get: returning cached lineage (len={len(self.lineage)})"
        )
        return list(self.lineage)

    cards = await self.parent.Layout.get_cards(
        context=context,
        parent_auth_retrieval_fn=parent_auth_retrieval_fn,
    )

    self.cards = cards

    lineage_cache = self._lineage_cache
    lineage_groups: list[list[DomoLineage_Link]] = []
    card_links: list[DomoLineage_Link] = []

    for card in cards:
        card_lineage = await card.Lineage.get(
            session=session,
            debug_api=debug_api,
            return_raw=return_raw,
            parent_auth=parent_auth,
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            max_depth=max_depth,
            lineage_cache=lineage_cache,
            context=context,
        )

        card_root = next(
            (
                link
                for link in card_lineage
                if str(link.id) == str(card.id) and link.type == "CARD"
            ),
            None,
        )

        card_link = DomoLineageLink_Card(
            id=str(card.id),
            auth=card.auth,
            entity=card,
            _type=None,
            dependencies=card_root.dependencies if card_root else [],
            dependents=card_root.dependents if card_root else [],
        )
        card_links.append(card_link)
        lineage_groups.append(card_lineage)

    page_link = self._build_root_link(
        dependencies=card_links,
        dependents=[],
    )

    merged_lineage = self._merge_lineage_links(
        [page_link],
        card_links,
        *lineage_groups,
    )

    self.immediate_dependencies = card_links
    self.immediate_dependents = []
    self.lineage = [page_link] + [
        link for link in merged_lineage if link is not page_link
    ]

    # Connect subscriber page → publisher page if a publisher PAGE
    # exists in the merged lineage (it comes from card federation chains).
    for link in self.lineage:
        if link.type == "PAGE" and str(link.id) != str(page_link.id):
            self._add_direct_link(page_link, link)
            break

    # Connect subscriber page → subscription so the diagram shows the
    # page directly linked to the subscription entity.
    for link in self.lineage:
        if link.type == "SUBSCRIPTION":
            self._add_direct_link(page_link, link)
            break

    self._cached_lineage_params = cache_key

    return list(self.lineage)

get_downstream `async` ¶

get_downstream(
    session: AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    max_depth: int = 1,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoLineage_Link]

Get downstream lineage for a page.

A page's downstream dependents are publications that include this page. Also includes any datacenter API results.

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

async def get_downstream(
    self,
    session: httpx.AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    max_depth: int = 1,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoLineage_Link]:
    """Get downstream lineage for a page.

    A page's downstream dependents are publications that include this page.
    Also includes any datacenter API results.
    """
    from ..DomoEverywhere.lineage import DomoLineageLink_Publication

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    # Try to find publications that include this page
    publication_links: list[DomoLineage_Link] = []
    try:
        # Check if the page is published via the Federation system
        if hasattr(self.parent, "Federation") and self.parent.Federation:
            federation = self.parent.Federation
            if hasattr(federation, "is_published") and federation.is_published:
                # Get the subscription/publication chain
                subscription = getattr(federation, "subscription", None)
                if subscription:
                    publication = getattr(federation, "parent_publication", None)
                    if publication:
                        pub_link = DomoLineageLink_Publication(
                            id=str(publication.id),
                            auth=self.auth,
                            entity=publication,
                            _type="PUBLICATION",
                            dependencies=[],
                            dependents=[],
                        )
                        publication_links.append(pub_link)
    except (AttributeError, ValueError, TypeError) as exc:
        await logger.warning(
            f"DomoLineage_Page.get_downstream: publication discovery failed for page {self.parent.id}: {exc}"
        )

    # Also try the base class datacenter API
    try:
        base_downstream = await super().get_downstream(
            session=session,
            debug_api=debug_api,
            return_raw=return_raw,
            max_depth=max_depth,
            context=context,
            **context_kwargs,
        )
    except (httpx.HTTPStatusError, ValueError) as exc:
        await logger.warning(
            f"DomoLineage_Page.get_downstream: base downstream lookup failed for page {self.parent.id}: {exc}"
        )
        base_downstream = []

    # Merge publication links with base downstream results
    all_links = publication_links + [
        link for link in base_downstream if link not in publication_links
    ]

    self.immediate_dependents = publication_links
    self.downstream_lineage = all_links

    return all_links

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