core

core

DomoEverywhere dataclass

DomoEverywhere(
    auth: DomoAuth,
    publications: list[DomoPublication] = None,
    subscriptions: list[DomoSubscription] = None,
    invitations: list[dict[str, Any]] = None,
)

Instance-wide Domo Everywhere access: publications, subscriptions, invites.

accept_invite_by_id async

accept_invite_by_id(
    subscription_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> Any

Accept a subscription invitation.

Parameters:

Name	Type	Description	Default
subscription_id	str	ID of the subscription invitation.	required
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
Any	The response payload on HTTP 200, otherwise None.

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

async def accept_invite_by_id(
    self,
    subscription_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> Any:
    """Accept a subscription invitation.

    Args:
        subscription_id: ID of the subscription invitation.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The response payload on HTTP 200, otherwise None.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    res = await publish_routes.accept_invite_by_id(
        auth=self.auth,
        subscription_id=subscription_id,
        context=context,
    )

    if res.status == 200:
        return res.response
    return None

count_subscription_invites async

count_subscription_invites(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Count subscription invites (/subscriptions/invites/counts).

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def count_subscription_invites(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Count subscription invites (``/subscriptions/invites/counts``).

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.count_subscription_invites(
        auth=self.auth,
        context=context,
    )

count_subscription_summaries async

count_subscription_summaries(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Count subscription summaries (/subscriptions/summaries/counts).

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def count_subscription_summaries(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Count subscription summaries (``/subscriptions/summaries/counts``).

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.count_subscription_summaries(
        auth=self.auth,
        context=context,
    )

get_publications async

get_publications(
    search_term: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[DomoPublication] | ResponseGetData

Get this instance's publications.

Parameters:

Name	Type	Description	Default
search_term	str | None	Optional text to filter publications by.	None
return_raw	bool	If True, return the raw search response.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
list[DomoPublication] | ResponseGetData	The publications (also stored on publications), or the raw
list[DomoPublication] | ResponseGetData	response when return_raw is True.

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

async def get_publications(
    self,
    search_term: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[DomoPublication] | rgd.ResponseGetData:
    """Get this instance's publications.

    Args:
        search_term: Optional text to filter publications by.
        return_raw: If True, return the raw search response.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The publications (also stored on ``publications``), or the raw
        response when ``return_raw`` is True.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    res = await publish_routes.search_publications(
        auth=self.auth,
        search_term=search_term,
        context=context,
    )

    if return_raw:
        return res

    self.publications = await dmce.gather_with_concurrency(
        n=60,
        *[  # noqa: B026
            DomoPublication.get_by_id(
                publication_id=obj.get("id"), auth=self.auth, context=context
            )
            for obj in res.response
        ],
    )
    return self.publications

get_subscription_invitations async

get_subscription_invitations(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get this instance's pending subscription invitations.

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response (its payload is also stored on invitations).

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

async def get_subscription_invitations(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get this instance's pending subscription invitations.

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response (its payload is also stored on ``invitations``).
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    res = await publish_routes.get_subscription_invitations(
        auth=self.auth,
        context=context,
    )

    self.invitations = res.response

    return res

get_subscription_invite_counts async

get_subscription_invite_counts(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Count subscription invites (/subscription/invites/counts).

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def get_subscription_invite_counts(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Count subscription invites (``/subscription/invites/counts``).

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.get_subscription_invite_counts(
        auth=self.auth,
        context=context,
    )

get_subscription_status async

get_subscription_status(
    package_publication_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get subscription status for a package publication.

Parameters:

Name	Type	Description	Default
package_publication_id	str	Package publication ID.	required
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def get_subscription_status(
    self,
    package_publication_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get subscription status for a package publication.

    Args:
        package_publication_id: Package publication ID.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.get_subscription_status(
        auth=self.auth,
        package_publication_id=package_publication_id,
        context=context,
    )

get_subscription_summaries_by_pub_id async

get_subscription_summaries_by_pub_id(
    pub_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get the subscription summaries of one publication.

Parameters:

Name	Type	Description	Default
pub_id	str	Publication ID.	required
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def get_subscription_summaries_by_pub_id(
    self,
    pub_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the subscription summaries of one publication.

    Args:
        pub_id: Publication ID.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.get_subscription_summaries_by_pub_id(
        auth=self.auth,
        pub_id=pub_id,
        context=context,
    )

get_subscription_summary_counts async

get_subscription_summary_counts(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Count subscription summaries (/subscription/summaries/counts).

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def get_subscription_summary_counts(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Count subscription summaries (``/subscription/summaries/counts``).

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.get_subscription_summary_counts(
        auth=self.auth,
        context=context,
    )

get_subscriptions async

get_subscriptions(
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[DomoSubscription] | ResponseGetData

Get this instance's subscription summaries.

Parameters:

Name	Type	Description	Default
return_raw	bool	If True, return the raw summaries response.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
list[DomoSubscription] | ResponseGetData	The subscriptions (also stored on subscriptions), or the raw
list[DomoSubscription] | ResponseGetData	response when return_raw is True.

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

async def get_subscriptions(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[DomoSubscription] | rgd.ResponseGetData:
    """Get this instance's subscription summaries.

    Args:
        return_raw: If True, return the raw summaries response.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The subscriptions (also stored on ``subscriptions``), or the raw
        response when ``return_raw`` is True.
    """

    self.subscriptions = []

    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    res = await publish_routes.get_subscription_summaries(
        auth=self.auth,
        context=context,
    )

    if return_raw:
        return res

    for sub in res.response:
        domo_sub = DomoSubscription.from_dict(auth=self.auth, obj=sub)

        if sub in self.subscriptions:
            continue
        self.subscriptions.append(domo_sub)

    return self.subscriptions

list_all_subscriptions async

list_all_subscriptions(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

List every subscription on this instance.

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def list_all_subscriptions(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List every subscription on this instance.

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.list_all_subscriptions(
        auth=self.auth,
        context=context,
    )

list_public_subscription_invites async

list_public_subscription_invites(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

List public subscription invites.

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def list_public_subscription_invites(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List public subscription invites.

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.list_public_subscription_invites(
        auth=self.auth,
        context=context,
    )

list_subscription_invites async

list_subscription_invites(
    limit: int | None = None,
    offset: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

List subscription invites.

Parameters:

Name	Type	Description	Default
limit	int | None	Optional page size.	None
offset	int | None	Optional page offset.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def list_subscription_invites(
    self,
    limit: int | None = None,
    offset: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List subscription invites.

    Args:
        limit: Optional page size.
        offset: Optional page offset.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.list_subscription_invites(
        auth=self.auth,
        limit=limit,
        offset=offset,
        context=context,
    )

list_subscription_summaries async

list_subscription_summaries(
    limit: int | None = None,
    offset: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

List subscription summaries.

Parameters:

Name	Type	Description	Default
limit	int | None	Optional page size.	None
offset	int | None	Optional page offset.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def list_subscription_summaries(
    self,
    limit: int | None = None,
    offset: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List subscription summaries.

    Args:
        limit: Optional page size.
        offset: Optional page offset.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    return await publish_subscription_routes.list_subscription_summaries(
        auth=self.auth,
        limit=limit,
        offset=offset,
        context=context,
    )

search_publications async

search_publications(
    search_term: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[DomoPublication] | ResponseGetData

Search this instance's publications (alias of get_publications).

Parameters:

Name	Type	Description	Default
search_term	str | None	Optional text to filter publications by.	None
return_raw	bool	If True, return the raw search response.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
list[DomoPublication] | ResponseGetData	The publications, or the raw response when return_raw is True.

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

async def search_publications(
    self,
    search_term: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[DomoPublication] | rgd.ResponseGetData:
    """Search this instance's publications (alias of ``get_publications``).

    Args:
        search_term: Optional text to filter publications by.
        return_raw: If True, return the raw search response.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The publications, or the raw response when ``return_raw`` is True.
    """
    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    res = await self.get_publications(
        search_term=search_term,
        return_raw=return_raw,
        context=context,
    )

    return res  # noqa: RET504

DomoPublication dataclass

DomoPublication(
    auth: DomoAuth,
    id: str,
    raw: dict,
    name: str = None,
    description: str = None,
    is_v2: bool = None,
    created_dt: datetime = None,
    updated_dt: datetime = None,
    subscriptions: list[DomoSubscription] = None,
    content: list[DomoPublication_Content] = None,
)

Bases: DomoEntity_w_Lineage

A Domo Everywhere publication: content a publisher sends to subscribers.

display_url property

display_url: str

URL of this publication's details page in the Domo admin UI.

create_publication async classmethod

create_publication(
    auth: DomoAuth,
    name: str,
    content_ls: list[DomoPublication_Content],
    subscription_ls: list[DomoSubscription],
    unique_id: str | None = None,
    description: str | None = None,
    debug_api: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoPublication

Create a publication.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session (the publisher).	required
name	str	Publication name.	required
content_ls	list[DomoPublication_Content]	Content to publish.	required
subscription_ls	list[DomoSubscription]	Subscribers to publish to.	required
unique_id	str | None	Optional publication ID; a UUID is generated if omitted.	None
description	str | None	Optional description.	None
debug_api	bool	Enable API debug logging.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
DomoPublication	The created publication.

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

@classmethod
async def create_publication(
    cls,
    auth: DomoAuth,
    name: str,
    content_ls: list[DomoPublication_Content],
    subscription_ls: list[DomoSubscription],
    unique_id: str | None = None,
    description: str | None = None,
    debug_api: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoPublication:
    """Create a publication.

    Args:
        auth: Authenticated Domo session (the publisher).
        name: Publication name.
        content_ls: Content to publish.
        subscription_ls: Subscribers to publish to.
        unique_id: Optional publication ID; a UUID is generated if omitted.
        description: Optional description.
        debug_api: Enable API debug logging.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The created publication.
    """
    if not isinstance(subscription_ls, list):
        subscription_ls = [subscription_ls]

    domain_ls = []
    content_json_ls = []
    for sub in subscription_ls:
        domain_ls.append(sub.subscriber_domain)
    for content_item in content_ls:
        content_json_ls.append(content_item.to_api_json())

    unique_id = unique_id or str(uuid.uuid4())

    body = publish_routes.generate_publish_body(
        url=f"{auth.domo_instance}.domo.com",
        sub_domain_ls=domain_ls,
        content_ls=content_json_ls,
        name=name,
        unique_id=unique_id,
        description=description or "",
        is_new=True,
    )

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

    res = await publish_routes.create_publish_job(
        auth=auth,
        body=body,
        context=context,
    )

    return cls.from_dict(auth=auth, obj=res.response)

from_dict classmethod

from_dict(
    auth: DomoAuth, obj: dict[str, Any]
) -> DomoPublication

Build a publication from an API payload.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
obj	dict[str, Any]	The publication payload.	required

Returns:

Type	Description
DomoPublication	The publication entity, with subscriptions and content populated.

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

@classmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any]) -> DomoPublication:
    """Build a publication from an API payload.

    Args:
        auth: Authenticated Domo session.
        obj: The publication payload.

    Returns:
        The publication entity, with subscriptions and content populated.
    """
    domo_pub = cls(
        id=obj["id"],
        name=obj["name"],
        description=obj["description"],
        created_dt=(
            dt.datetime.fromtimestamp(obj["created"] / 1000)
            if obj["created"]
            else None
        ),
        updated_dt=(
            dt.datetime.fromtimestamp(obj.get("content").get("updated") / 1000)
            if obj.get("content").get("updated")
            else None
        ),
        is_v2=obj["isV2"],
        auth=auth,
        raw=obj,
    )

    if (
        obj.get("subscriptionAuthorizations")
        and len(obj.get("subscriptionAuthorizations")) > 0
    ):
        domo_pub._generate_subscriptions(
            subscription_authorizations_ls=obj["subscriptionAuthorizations"],
            auth=auth,
        )

    if obj.get("children") and len(obj.get("children")) > 0:
        domo_pub._generate_content(obj["children"])

    return domo_pub

get_by_id async classmethod

get_by_id(
    publication_id: str,
    auth: DomoAuth,
    return_raw: bool = False,
    timeout: int = 10,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoPublication | ResponseGetData

Get a publication by ID.

Parameters:

Name	Type	Description	Default
publication_id	str	Publication ID.	required
auth	DomoAuth	Authenticated Domo session.	required
return_raw	bool	If True, return the raw response instead of an entity.	False
timeout	int	Request timeout in seconds.	10
debug_api	bool	Enable API debug logging.	False
debug_num_stacks_to_drop	int	Stack frames to drop in debug output.	2
session	AsyncClient | None	Optional shared HTTP client.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
DomoPublication | ResponseGetData	The publication, or the raw response when return_raw is True.

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

@classmethod
async def get_by_id(
    cls,
    publication_id: str,
    auth: DomoAuth,
    return_raw: bool = False,
    timeout: int = 10,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoPublication | rgd.ResponseGetData:
    """Get a publication by ID.

    Args:
        publication_id: Publication ID.
        auth: Authenticated Domo session.
        return_raw: If True, return the raw response instead of an entity.
        timeout: Request timeout in seconds.
        debug_api: Enable API debug logging.
        debug_num_stacks_to_drop: Stack frames to drop in debug output.
        session: Optional shared HTTP client.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The publication, or the raw response when ``return_raw`` is True.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await publish_routes.get_publication_by_id(
        auth=auth,
        publication_id=publication_id,
        timeout=timeout,
        context=context,
    )

    if return_raw:
        return res

    return cls.from_dict(auth=auth, obj=res.response)

get_content_details async

get_content_details(
    subscriber_domain: str,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Map this publication's content to the subscriber's copies.

Sets subscriber_content_id and subscriber_insance on each content item.

Parameters:

Name	Type	Description	Default
subscriber_domain	str	Subscriber instance domain; .domo.com is appended if missing.	required
debug_api	bool	Enable API debug logging.	False
debug_num_stacks_to_drop	int	Stack frames to drop in debug output.	2
session	AsyncClient | None	Optional shared HTTP client.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The content-details response.

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

async def get_content_details(
    self,
    subscriber_domain: str,  # must include .domo.com
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Map this publication's content to the subscriber's copies.

    Sets ``subscriber_content_id`` and ``subscriber_insance`` on each content item.

    Args:
        subscriber_domain: Subscriber instance domain; ``.domo.com`` is appended if missing.
        debug_api: Enable API debug logging.
        debug_num_stacks_to_drop: Stack frames to drop in debug output.
        session: Optional shared HTTP client.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The content-details response.
    """
    if not subscriber_domain.lower().endswith(".domo.com"):
        subscriber_domain = f"{subscriber_domain}.domo.com"

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

    res = await publish_routes.get_subscriber_content_details(
        auth=self.auth,
        publication_id=self.id,
        subscriber_instance=subscriber_domain,
        context=context,
    )

    publication_content = self.content

    for content in publication_content:
        subscriber_obj = next(
            (
                subscriber_obj
                for subscriber_obj in res.response
                if subscriber_obj["publisherObjectId"] == content.entity_id
                and subscriber_obj["contentType"] == content.entity_type
            ),
            None,
        )
        if subscriber_obj is not None:
            content.subscriber_content_id = subscriber_obj["subscriberObjectId"]
            content.subscriber_insance = subscriber_obj["subscriberDomain"]

    self.content = publication_content

    return res

get_entity_by_id async classmethod

get_entity_by_id(
    entity_id: str, **kwargs: Any
) -> DomoPublication | ResponseGetData

Get a publication by ID (generic entity entry point).

Parameters:

Name	Type	Description	Default
entity_id	str	Publication ID.	required
**kwargs	Any	Forwarded to get_by_id.	{}

Returns:

Type	Description
DomoPublication | ResponseGetData	The publication, or the raw response when return_raw is True.

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

@classmethod
async def get_entity_by_id(
    cls, entity_id: str, **kwargs: Any
) -> DomoPublication | rgd.ResponseGetData:
    """Get a publication by ID (generic entity entry point).

    Args:
        entity_id: Publication ID.
        **kwargs: Forwarded to ``get_by_id``.

    Returns:
        The publication, or the raw response when ``return_raw`` is True.
    """
    return await cls.get_by_id(publication_id=entity_id, **kwargs)

get_publisher_entity_for_subscriber async

get_publisher_entity_for_subscriber(
    subscriber_entity: Any,
    subscriber_domain: str,
    *,
    include_indirect: bool = True,
    context: RouteContext | None = None,
    **context_kwargs
) -> Any

Get publisher entity with automatic direct/indirect resolution.

Handles: - Direct publication (entity in publication content) - Indirect publication (cards on published pages) - Multi-page publications - App Studio caveats

Parameters:

Name	Type	Description	Default
subscriber_entity	Any	Subscriber-side entity	required
subscriber_domain	str	Subscriber instance domain	required
include_indirect	bool	Enable indirect resolution (cards on pages)	True
context	RouteContext | None	Route context	None
**context_kwargs		Additional context args	{}

Returns:

Type	Description
Any	Publisher entity or None if not found

Raises:

Type	Description
NotImplementedError	If App Studio indirect resolution attempted

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

async def get_publisher_entity_for_subscriber(  # noqa: C901
    self,
    subscriber_entity: Any,  # DomoEntity_w_Lineage
    subscriber_domain: str,
    *,
    include_indirect: bool = True,
    context: RouteContext | None = None,
    **context_kwargs,
) -> Any:  # DomoCard, DomoDataset, DomoPage
    """Get publisher entity with automatic direct/indirect resolution.

    Handles:
    - Direct publication (entity in publication content)
    - Indirect publication (cards on published pages)
    - Multi-page publications
    - App Studio caveats

    Args:
        subscriber_entity: Subscriber-side entity
        subscriber_domain: Subscriber instance domain
        include_indirect: Enable indirect resolution (cards on pages)
        context: Route context
        **context_kwargs: Additional context args

    Returns:
        Publisher entity or None if not found

    Raises:
        NotImplementedError: If App Studio indirect resolution attempted
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    # Check cache first
    cache_key = f"{subscriber_entity.entity_type}:{subscriber_entity.id}"
    cache = self.raw.setdefault("_indirect_resolution_cache", {})

    if cache_key in cache:
        return cache[cache_key]

    # Get content mapping
    res = await self.get_content_details(
        subscriber_domain=subscriber_domain,
        context=context,
    )

    # Try direct match (normalize dataset content type aliases)
    normalized_entity_type = (subscriber_entity.entity_type or "").upper()
    acceptable_content_types = {normalized_entity_type}
    # entity_type is now consistently "DATA_SOURCE" (not "DATASET"),
    # but the API may return either value, so accept both aliases
    if normalized_entity_type in ("DATA_SOURCE", "DATASET"):
        acceptable_content_types.update({"DATA_SOURCE", "DATASET"})

    direct_match = next(
        (
            row
            for row in res.response
            if str(row["subscriberObjectId"]) == str(subscriber_entity.id)
            and (row.get("contentType") or "").upper() in acceptable_content_types
        ),
        None,
    )

    if direct_match:
        # Direct publication - simple case
        publisher_obj_id = direct_match["publisherObjectId"]

        publisher_entity = await subscriber_entity.get_entity_by_id(
            entity_id=publisher_obj_id, auth=self.auth
        )

        cache[cache_key] = publisher_entity
        return publisher_entity

    # No direct match - try indirect resolution if enabled
    if not include_indirect:
        return None

    if subscriber_entity.entity_type != "CARD":
        # Only cards can be indirectly published (via pages)
        return None

    # Find PAGE entries in publication
    page_contents = [c for c in self.content if c.entity_type == "PAGE"]

    if not page_contents:
        # Check for App Studio content
        app_contents = [
            c for c in self.content if c.entity_type in {"APP", "APP_STUDIO_APP"}
        ]
        if app_contents:
            # TODO(#XXX): Implement App Studio card resolution
            raise NotImplementedError(
                f"Indirect publication resolution for App Studio cards is not yet supported.\n"
                f"Card: {subscriber_entity.id}\n"
                f"App Studio content: {[f'{c.entity_type}:{c.entity_id}' for c in app_contents]}\n"
                f"See GitHub issue for tracking."
            )
        return None

    # Prefer mapping by subscriber page IDs when resolving cards
    from ..DomoPage import DomoPage

    subscriber_page_ids: list[str] = []
    if subscriber_entity.Lineage:
        try:
            dependents = await subscriber_entity.Lineage._get_immediate_dependents(
                context=context
            )
            for link in dependents:
                if link.type != "PAGE":
                    continue
                if link.entity is None:
                    link.entity = await link.get_entity(context=context)
                if link.entity:
                    subscriber_page_ids.append(str(link.entity.id))
        except dmde.DomoError:
            subscriber_page_ids = []

    # If we know subscriber pages, narrow to matching published pages
    page_mappings = []
    if subscriber_page_ids:
        page_mappings = [
            row
            for row in res.response
            if row.get("contentType") == "PAGE"
            and str(row.get("subscriberObjectId")) in subscriber_page_ids
        ]

    if page_mappings:
        for row in page_mappings:
            publisher_page = await DomoPage.get_by_id(
                auth=self.auth,
                page_id=row.get("publisherObjectId"),
                context=context,
            )

            publisher_cards = await publisher_page.Layout.get_cards(context=context)

            if not publisher_cards:
                continue

            # Store cards on page for graph building (enables Page→Card edges)
            if publisher_page.Layout:
                publisher_page.Layout.cards = publisher_cards

            # Store page entity in content for graph building
            page_content_match = next(
                (
                    c
                    for c in page_contents
                    if str(c.entity_id) == str(row.get("publisherObjectId"))
                ),
                None,
            )
            if page_content_match:
                page_content_match.entity = publisher_page

            # Match by title (reliable for cards)
            matching_card = next(
                (c for c in publisher_cards if c.title == subscriber_entity.title),
                None,
            )

            if matching_card:
                cache[cache_key] = matching_card
                return matching_card

            # Fallback: single-card page heuristic
            if len(publisher_cards) == 1:
                cache[cache_key] = publisher_cards[0]
                return publisher_cards[0]

    # Fallback: Search each published page for the card (legacy heuristic)
    for page_content in page_contents:
        publisher_page = await DomoPage.get_by_id(
            auth=self.auth,
            page_id=page_content.entity_id,
            context=context,
        )

        publisher_cards = await publisher_page.Layout.get_cards(context=context)

        if not publisher_cards:
            continue

        # Store cards on page for graph building (enables Page→Card edges)
        if publisher_page.Layout:
            publisher_page.Layout.cards = publisher_cards

        # Store page entity in content for graph building
        page_content.entity = publisher_page

        # Match by title (reliable for cards)
        matching_card = next(
            (c for c in publisher_cards if c.title == subscriber_entity.title),
            None,
        )

        if matching_card:
            cache[cache_key] = matching_card
            return matching_card

        # Fallback: single-card page heuristic
        if len(publisher_cards) == 1:
            cache[cache_key] = publisher_cards[0]
            return publisher_cards[0]

    # No matching card found
    return None

revoke_subscription_auth async

revoke_subscription_auth(
    auth: DomoAuth = None,
    subscription_id: str | None = None,
    subscription: DomoSubscription | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> None

Revoke a subscription's authorization (not yet supported).

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session; defaults to this publication's.	None
subscription_id	str | None	Subscription ID; defaults to subscription.id.	None
subscription	DomoSubscription | None	Subscription to revoke.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Raises:

Type	Description
NotImplementedError	Always; no revoke route exists yet.

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

async def revoke_subscription_auth(
    self,
    auth: DomoAuth = None,
    subscription_id: str | None = None,
    subscription: DomoSubscription | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> None:
    """Revoke a subscription's authorization (not yet supported).

    Args:
        auth: Authenticated Domo session; defaults to this publication's.
        subscription_id: Subscription ID; defaults to ``subscription.id``.
        subscription: Subscription to revoke.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Raises:
        NotImplementedError: Always; no revoke route exists yet.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    # Note: This function may not have a direct route implementation
    # It would need to be implemented based on the available API endpoints
    auth = auth or self.auth
    subscription_id = subscription_id or subscription.id

    # This is a placeholder implementation - would need actual API route
    raise NotImplementedError("revoke_subscription_auth route not yet implemented")

update_publication async

update_publication(
    content_ls: list[DomoPublication_Content] | None = None,
    description: str | None = None,
    name: str | None = None,
    subscription_ls: list[DomoSubscription] | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Update this publication.

Parameters:

Name	Type	Description	Default
content_ls	list[DomoPublication_Content] | None	New content list.	None
description	str | None	New description; keeps the current one if omitted.	None
name	str | None	New name; keeps the current one if omitted.	None
subscription_ls	list[DomoSubscription] | None	New subscriber list.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The update response.

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

async def update_publication(
    self,
    content_ls: list[DomoPublication_Content] | None = None,
    description: str | None = None,
    name: str | None = None,
    subscription_ls: list[DomoSubscription] | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Update this publication.

    Args:
        content_ls: New content list.
        description: New description; keeps the current one if omitted.
        name: New name; keeps the current one if omitted.
        subscription_ls: New subscriber list.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The update response.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    # Use the actual available route
    if not isinstance(subscription_ls, list) and subscription_ls:
        subscription_ls = [subscription_ls]

    domain_ls = []
    content_json_ls = []

    if subscription_ls:
        for sub in subscription_ls:
            domain_ls.append(sub.subscriber_domain)

    if content_ls:
        for content_item in content_ls:
            content_json_ls.append(content_item.to_api_json())

    body = publish_routes.generate_publish_body(
        url=f"{self.auth.domo_instance}.domo.com",
        sub_domain_ls=domain_ls,
        content_ls=content_json_ls,
        name=name or self.name,
        unique_id=self.id,
        description=description or self.description,
        is_new=False,
    )

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

    res = await publish_routes.update_publish_job(
        auth=self.auth,
        publication_id=self.id,
        body=body,
        context=context,
    )

    return res  # noqa: RET504

DomoPublication_Content dataclass

DomoPublication_Content(
    auth: DomoAuth,
    content_id: str,
    entity_type: str,
    entity_id: str,
    entity_domain: str,
    is_v2: bool,
    is_direct_content: bool,
    created_dt: datetime,
    updated_dt: datetime = None,
    subscriber_content_id: str = None,
    subscriber_insance: str = None,
    entity: Any = None,
    parent: Any = None,
)

A piece of publisher content (card, dataset, page, data app) in a publication.

from_dict classmethod

from_dict(
    auth: DomoAuth, obj: dict[str, Any], parent: Any = None
) -> DomoPublication_Content

Build publication content from a publication children entry.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
obj	dict[str, Any]	One entry of the publication's children list.	required
parent	Any	The owning publication, if known.	None

Returns:

Type	Description
DomoPublication_Content	The publication content entity.

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

@classmethod
def from_dict(
    cls, auth: DomoAuth, obj: dict[str, Any], parent: Any = None
) -> DomoPublication_Content:
    """Build publication content from a publication ``children`` entry.

    Args:
        auth: Authenticated Domo session.
        obj: One entry of the publication's ``children`` list.
        parent: The owning publication, if known.

    Returns:
        The publication content entity.
    """
    entity_type = obj.get("content").get("type")
    return cls(
        auth=auth,
        content_id=obj["id"],
        entity_id=obj.get("content").get("domoObjectId"),
        entity_domain=obj.get("content").get("domain"),
        is_v2=obj.get("isV2"),
        created_dt=(
            dt.datetime.fromtimestamp(obj["created"] / 1000)
            if obj["created"]
            else None
        ),
        updated_dt=(
            dt.datetime.fromtimestamp(obj.get("content").get("updated") / 1000)
            if obj.get("content").get("updated")
            else None
        ),
        is_direct_content=obj.get("useDirectContent"),
        parent=parent,
        entity_type=entity_type,
        entity=DomoPublication_Content_Enum[entity_type].value,
    )

get_entity async

get_entity(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> Any

Get the entity behind this publication content.

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
Any	The publisher-side entity (card, dataset, page, or data app).

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

async def get_entity(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> Any:
    """Get the entity behind this publication content.

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The publisher-side entity (card, dataset, page, or data app).
    """
    if not self.entity:
        self.entity = DomoPublication_Content_Enum[self.entity_type].value

    base_context = RouteContext.build_context(
        parent_class=self.__class__.__name__,
        **context_kwargs,
    )
    context = RouteContext.build_context(context=context or base_context)

    self.entity = await self.entity.get_entity_by_id(
        auth=self.auth,
        entity_id=self.entity_id,
        context=context,
    )

    return self.entity

to_api_json

to_api_json() -> dict[str, Any]

Serialize this content for a publish create/update request body.

Returns:

Type	Description
dict[str, Any]	The content descriptor expected by the publish API.

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

def to_api_json(self) -> dict[str, Any]:
    """Serialize this content for a publish create/update request body.

    Returns:
        The content descriptor expected by the publish API.
    """
    return {
        "domain": self.entity_domain,
        "domoObjectId": self.entity_id,
        "customerId": self.entity_domain,
        "type": self.entity_type,
    }

DomoPublication_Content_Enum

Bases: DomoEnumMixin, Enum

Maps a publication content type name to the entity class that models it.

DomoPublication_UnexpectedContentType

DomoPublication_UnexpectedContentType(
    cls_instance: Any, content_type: str
)

Bases: ClassError

Raised when a publication holds a content type the library does not model.

Build the error for an unexpected content type.

Parameters:

Name	Type	Description	Default
cls_instance	Any	The publication that holds the content.	required
content_type	str	The unrecognized content type name.	required

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

def __init__(self, cls_instance: Any, content_type: str) -> None:
    """Build the error for an unexpected content type.

    Args:
        cls_instance: The publication that holds the content.
        content_type: The unrecognized content type name.
    """
    super().__init__(
        cls_instance=cls_instance,
        message=f"DomoPublication_Instantiation: Unexpected content type {content_type}",
    )

DomoSubscription dataclass

DomoSubscription(
    auth: DomoAuth,
    id: str,
    raw: dict,
    publication_id: str,
    subscriber_domain: str,
    publisher_domain: str,
    parent_publication: DomoPublication = None,
    created_dt: datetime | None = None,
    _name: str | None = None,
)

Bases: DomoEntity_w_Lineage

A subscription: an instance that a publication is sent to.

display_url property

display_url: str

URL of the subscriptions page in the Domo admin UI.

name property

name: str

Get subscription name.

Priority: 1. Name from API response (if available) 2. Parent publication name (no extra API call if already loaded) 3. Fallback to "Subscription"

from_dict classmethod

from_dict(
    auth: DomoAuth,
    obj: dict[str, Any],
    parent_publication: Any = None,
) -> DomoSubscription

Build a subscription from an API payload.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
obj	dict[str, Any]	The subscription payload.	required
parent_publication	Any	The owning publication, if known.	None

Returns:

Type	Description
DomoSubscription	The subscription entity.

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

@classmethod
def from_dict(
    cls, auth: DomoAuth, obj: dict[str, Any], parent_publication: Any = None
) -> DomoSubscription:
    """Build a subscription from an API payload.

    Args:
        auth: Authenticated Domo session.
        obj: The subscription payload.
        parent_publication: The owning publication, if known.

    Returns:
        The subscription entity.
    """
    return cls(
        auth=auth,
        id=obj.get("id") or obj.get("subscriptionId"),
        publication_id=obj["publicationId"],
        subscriber_domain=obj.get("domain") or obj.get("subscriberDomain"),
        publisher_domain=obj.get("publisherDomain"),
        created_dt=(
            (dt.datetime.fromtimestamp(obj.get("created") / 1000))
            if obj.get("created")
            else None
        ),
        _name=obj.get("name"),  # Extract name from API if available
        raw=obj,
        parent_publication=parent_publication,
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    subscription_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoSubscription | ResponseGetData

Get a subscription by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
subscription_id	str	Subscription ID.	required
return_raw	bool	If True, return the raw response instead of an entity.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
DomoSubscription | ResponseGetData	The subscription, or the raw response when return_raw is True.

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

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    subscription_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoSubscription | rgd.ResponseGetData:
    """Get a subscription by ID.

    Args:
        auth: Authenticated Domo session.
        subscription_id: Subscription ID.
        return_raw: If True, return the raw response instead of an entity.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The subscription, or the raw response when ``return_raw`` is True.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await publish_routes.get_subscription_by_id(
        auth=auth,
        subscription_id=subscription_id,
        context=context,
    )

    if return_raw:
        return res

    return cls.from_dict(auth=auth, obj=res.response)

get_content_details async

get_content_details(
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[DomoPublication_Content]

Map the parent publication's content to this subscriber's copies.

Parameters:

Name	Type	Description	Default
parent_auth	DomoAuth | None	Auth for the publisher instance.	None
parent_auth_retrieval_fn	Callable | None	Callable returning publisher auth given the publisher domain; used when parent_auth is omitted.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
list[DomoPublication_Content]	The publication content, updated with subscriber-side IDs.

Raises:

Type	Description
DomoSubscription_NoParent	If the parent publication cannot be retrieved.

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

async def get_content_details(
    self,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[DomoPublication_Content]:
    """Map the parent publication's content to this subscriber's copies.

    Args:
        parent_auth: Auth for the publisher instance.
        parent_auth_retrieval_fn: Callable returning publisher auth given the
            publisher domain; used when ``parent_auth`` is omitted.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The publication content, updated with subscriber-side IDs.

    Raises:
        DomoSubscription_NoParent: If the parent publication cannot be retrieved.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not self.parent_publication:
        await self.get_parent_publication(
            parent_auth=parent_auth,
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            context=context,
        )

    if not self.parent_publication:
        raise DomoSubscription_NoParent(self)

    publication_content = self.parent_publication.content

    res = await publish_routes.get_subscriber_content_details(
        auth=self.parent_publication.auth,
        publication_id=self.publication_id,
        subscriber_instance=self.subscriber_domain,
        context=context,
    )

    for content in publication_content:
        subscriber_obj = next(
            (
                subscriber_obj
                for subscriber_obj in res.response
                if subscriber_obj["publisherObjectId"] == content.entity_id
                and subscriber_obj["contentType"] == content.entity_type
            ),
            None,
        )
        if subscriber_obj is not None:
            content.subscriber_content_id = subscriber_obj["subscriberObjectId"]
            content.subscriber_insance = subscriber_obj["subscriberDomain"]

    # Return the updated publication content
    return publication_content

get_entity_by_id async classmethod

get_entity_by_id(
    entity_id: str, auth: DomoAuth, **kwargs: Any
) -> DomoSubscription | ResponseGetData

Get a subscription by ID (generic entity entry point).

Parameters:

Name	Type	Description	Default
entity_id	str	Subscription ID.	required
auth	DomoAuth	Authenticated Domo session.	required
**kwargs	Any	Forwarded to get_by_id.	{}

Returns:

Type	Description
DomoSubscription | ResponseGetData	The subscription, or the raw response when return_raw is True.

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

@classmethod
async def get_entity_by_id(
    cls, entity_id: str, auth: DomoAuth, **kwargs: Any
) -> DomoSubscription | rgd.ResponseGetData:
    """Get a subscription by ID (generic entity entry point).

    Args:
        entity_id: Subscription ID.
        auth: Authenticated Domo session.
        **kwargs: Forwarded to ``get_by_id``.

    Returns:
        The subscription, or the raw response when ``return_raw`` is True.
    """
    return await cls.get_by_id(subscription_id=entity_id, auth=auth, **kwargs)

get_parent_publication async

get_parent_publication(
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoPublication

Get the publication this subscription belongs to.

Parameters:

Name	Type	Description	Default
parent_auth	DomoAuth | None	Auth for the publisher instance.	None
parent_auth_retrieval_fn	Callable | None	Callable returning publisher auth given the publisher domain (sync or async); used when parent_auth is omitted.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
DomoPublication	The parent publication (also stored on parent_publication).

Raises:

Type	Description
DomoSubscription_NoParentAuth	If no publisher auth can be resolved.

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

async def get_parent_publication(
    self,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoPublication:
    """Get the publication this subscription belongs to.

    Args:
        parent_auth: Auth for the publisher instance.
        parent_auth_retrieval_fn: Callable returning publisher auth given the
            publisher domain (sync or async); used when ``parent_auth`` is omitted.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The parent publication (also stored on ``parent_publication``).

    Raises:
        DomoSubscription_NoParentAuth: If no publisher auth can be resolved.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    if not parent_auth and parent_auth_retrieval_fn:
        auth_or_coro = parent_auth_retrieval_fn(self.publisher_domain)
        if inspect.isawaitable(auth_or_coro):
            parent_auth = await auth_or_coro  # type: ignore[assignment]
        else:
            parent_auth = auth_or_coro

    if not parent_auth:
        raise DomoSubscription_NoParentAuth(self)

    self.parent_publication = await DomoPublication.get_by_id(
        publication_id=self.publication_id,
        auth=parent_auth,
        context=context,
    )

    return self.parent_publication

get_share async

get_share(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get this subscription's share.

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def get_share(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get this subscription's share.

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    return await publish_subscription_routes.get_subscription_share(
        auth=self.auth,
        subscription_id=self.id,
        context=context,
    )

get_share_groups async

get_share_groups(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get the groups this subscription is shared with.

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def get_share_groups(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the groups this subscription is shared with.

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    return await publish_subscription_routes.get_subscription_share_groups(
        auth=self.auth,
        subscription_id=self.id,
        context=context,
    )

get_with_share async

get_with_share(
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get this subscription together with its share.

Parameters:

Name	Type	Description	Default
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

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

async def get_with_share(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get this subscription together with its share.

    Args:
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    return await publish_subscription_routes.get_subscription_with_share(
        auth=self.auth,
        subscription_id=self.id,
        context=context,
    )

DomoSubscription_NoParent

DomoSubscription_NoParent(cls_instance: Any)

Bases: ClassError

Raised when a subscription's parent publication cannot be retrieved.

Build the error for a subscription with no retrievable parent.

Parameters:

Name	Type	Description	Default
cls_instance	Any	The subscription whose parent publication is missing.	required

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

def __init__(self, cls_instance: Any) -> None:
    """Build the error for a subscription with no retrievable parent.

    Args:
        cls_instance: The subscription whose parent publication is missing.
    """
    super().__init__(
        cls_instance=cls_instance,
        entity_id="subscription_id",
        message="unable to retrieve parent publication",
    )

DomoSubscription_NoParentAuth

DomoSubscription_NoParentAuth(cls_instance: Any)

Bases: ClassError

Raised when a subscription has no parent-instance auth to reach its publication.

Build the error for a subscription that lacks parent auth.

Parameters:

Name	Type	Description	Default
cls_instance	Any	The subscription that needs parent auth.	required

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

def __init__(self, cls_instance: Any) -> None:
    """Build the error for a subscription that lacks parent auth.

    Args:
        cls_instance: The subscription that needs parent auth.
    """
    super().__init__(
        cls_instance=cls_instance,
        entity_id="subscription_id",
        message="must pass parent_auth or parent_auth_retrieval_fn which returns an instance of auth given self",
    )