DomoBuzz

DomoBuzz

Domo Buzz entity classes.

Wraps crew_dcs.routes.buzz (the Buzz product API — X-DOMO-Developer-Token on https://{instance}.domo.com/api, not the deprecated api.domo.com OAuth "Integration API").

Shapes to know before touching this module:

• GET /buzz/v1/channels/me returns a map keyed by channel id, not a list — each entry is a thin ChannelMeta (unread/important counts, mentioned, channelType, lastViewedTimestamp), not a full channel. :class:DomoBuzzChannelMeta models that shape; :class:DomoBuzzChannel models the richer object returned by GET /buzz/v1/channels, GET /buzz/v1/channels/{id}, and POST /buzz/v1/topics.
• POST /buzz/v1/sockets/authenticate is a standard Pusher authorizer — form-urlencoded request, JSON {"auth": ...} response. See :meth:DomoBuzzChannels.authenticate_socket and routes.buzz.authenticate_socket's docstring for the live-verified shape (an earlier "returns a bare string" claim was wrong and unverified).

DomoBuzzChannel dataclass

DomoBuzzChannel(
    auth: DomoAuth,
    id: str,
    raw: dict = dict(),
    channel_id: str | None = None,
    title: str | None = None,
    description: str | None = None,
    channel_type: str | None = None,
    locked: bool | None = None,
    created: int | None = None,
    created_by_id: str | None = None,
    created_by_type: str | None = None,
    modified: int | None = None,
    modified_by_id: str | None = None,
    modified_by_type: str | None = None,
    last_updated: int | None = None,
    permission_count: int | None = None,
    pinned_count: int | None = None,
    reply_count: int | None = None,
    default_permission: str | None = None,
    public_permission: str | None = None,
    via_id: str | None = None,
    via_type: str | None = None,
)

Bases: DomoEntity

A Buzz channel (topic, group chat, direct chat, or object chat).

Built from the rich channel payload returned by GET /buzz/v1/channels, GET /buzz/v1/channels/{id}, and POST /buzz/v1/topics (nested under response["channel"] for the latter). This is NOT the shape returned by GET /buzz/v1/channels/me — see :class:DomoBuzzChannelMeta for that.

create async classmethod

create(
    auth: DomoAuth,
    title: str,
    description: str | None = None,
    accesses: list[dict[str, Any]] | None = None,
    default_permission: str = "INVITE_OTHERS",
    public_permission: str = "INVITE_OTHERS",
    filters: list[dict[str, Any]] | None = None,
    pin: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoBuzzChannel | ResponseGetData

Create a Buzz topic and return it as a :class:DomoBuzzChannel.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

@classmethod
async def create(
    cls,
    auth: DomoAuth,
    title: str,
    description: str | None = None,
    accesses: list[dict[str, Any]] | None = None,
    default_permission: str = "INVITE_OTHERS",
    public_permission: str = "INVITE_OTHERS",
    filters: list[dict[str, Any]] | None = None,
    pin: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoBuzzChannel | ResponseGetData:
    """Create a Buzz topic and return it as a :class:`DomoBuzzChannel`."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await buzz_routes.create_topic(
        auth=auth,
        title=title,
        description=description,
        accesses=accesses,
        default_permission=default_permission,
        public_permission=public_permission,
        filters=filters,
        pin=pin,
        context=context,
    )

    if return_raw:
        return res

    payload = (
        res.response.get("channel")
        if isinstance(res.response, dict)
        else res.response
    )
    return cls.from_dict(auth=auth, obj=payload or {})

delete async

delete(
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool | ResponseGetData

Delete this Buzz channel.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def delete(
    self,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool | ResponseGetData:
    """Delete this Buzz channel."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await buzz_routes.delete_channel(
        auth=self.auth,
        channel_id=self.id,
        context=context,
    )

    if return_raw:
        return res

    return res.is_success

from_dict classmethod

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

Tolerant of a partial channel payload — only id is assumed.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

@classmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any]) -> DomoBuzzChannel:
    """Tolerant of a partial channel payload — only ``id`` is assumed."""
    return cls(
        auth=auth,
        id=str(obj.get("id") or obj.get("channelId") or ""),
        channel_id=obj.get("channelId"),
        title=obj.get("title"),
        description=obj.get("description"),
        channel_type=obj.get("channelType") or obj.get("type"),
        locked=obj.get("locked"),
        created=obj.get("created"),
        created_by_id=obj.get("createdById"),
        created_by_type=obj.get("createdByType"),
        modified=obj.get("modified"),
        modified_by_id=obj.get("modifiedById"),
        modified_by_type=obj.get("modifiedByType"),
        last_updated=obj.get("lastUpdated"),
        permission_count=obj.get("permissionCount"),
        pinned_count=obj.get("pinnedCount"),
        reply_count=obj.get("replyCount"),
        default_permission=obj.get("defaultPermission"),
        public_permission=obj.get("publicPermission"),
        via_id=obj.get("viaId"),
        via_type=obj.get("viaType"),
        raw=obj,
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    channel_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoBuzzChannel | ResponseGetData

Retrieve a Buzz channel by id.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    channel_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoBuzzChannel | ResponseGetData:
    """Retrieve a Buzz channel by id."""
    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 buzz_routes.get_channel_by_id(
        auth=auth,
        channel_id=channel_id,
        context=context,
    )

    if return_raw:
        return res

    payload = (
        res.response.get("channel")
        if isinstance(res.response, dict) and "channel" in res.response
        else res.response
    )
    return cls.from_dict(auth=auth, obj=payload or {})

get_entity_by_id async classmethod

get_entity_by_id(
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Any | None = None,
    parent_auth: Any | None = None,
    **kwargs
) -> DomoBuzzChannel

Internal wrapper used by the base refresh() method.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

@classmethod
async def get_entity_by_id(
    cls,
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Any | None = None,
    parent_auth: Any | None = None,
    **kwargs,
) -> DomoBuzzChannel:
    """Internal wrapper used by the base ``refresh()`` method."""
    return await cls.get_by_id(
        auth=auth,
        channel_id=entity_id,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        debug_api=debug_api,
        session=session,
    )

get_messages async

get_messages(
    paging_mode: str | None = None,
    paging_id: str | None = None,
    paging_field: str | None = None,
    paging_limit: int | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **kwargs
) -> list[DomoBuzzMessage] | ResponseGetData

Get message history for this channel.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def get_messages(
    self,
    paging_mode: str | None = None,
    paging_id: str | None = None,
    paging_field: str | None = None,
    paging_limit: int | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **kwargs,
) -> list[DomoBuzzMessage] | ResponseGetData:
    """Get message history for this channel."""
    context = RouteContext.build_context(context=context)

    res = await buzz_routes.get_channel_messages(
        auth=self.auth,
        channel_id=self.id,
        paging_mode=paging_mode,
        paging_id=paging_id,
        paging_field=paging_field,
        paging_limit=paging_limit,
        context=context,
        **kwargs,
    )

    if return_raw:
        return res

    messages = (
        res.response.get("messages", []) if isinstance(res.response, dict) else []
    )
    return [
        DomoBuzzMessage.from_dict(channel_id=self.id, obj=msg) for msg in messages
    ]

post_message async

post_message(
    content: str,
    guid: str | None = None,
    attachments: list[dict[str, Any]] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoBuzzMessage | ResponseGetData

Post a message to this channel (the richer, per-channel form).

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def post_message(
    self,
    content: str,
    guid: str | None = None,
    attachments: list[dict[str, Any]] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoBuzzMessage | ResponseGetData:
    """Post a message to this channel (the richer, per-channel form)."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await buzz_routes.create_channel_message(
        auth=self.auth,
        channel_id=self.id,
        content=content,
        guid=guid,
        attachments=attachments,
        context=context,
    )

    if return_raw:
        return res

    payload = (
        res.response.get("message")
        if isinstance(res.response, dict)
        else res.response
    )
    return DomoBuzzMessage.from_dict(channel_id=self.id, obj=payload or {})

DomoBuzzChannelMeta dataclass

DomoBuzzChannelMeta(
    id: str,
    raw: dict = dict(),
    channel_type: str | None = None,
    mentioned: bool | None = None,
    last_viewed_timestamp: int | None = None,
    unread_counts: dict | None = None,
    important_counts: dict | None = None,
)

One entry from GET /buzz/v1/channels/me's id-keyed map.

channels/me returns {"me": {<channelId>: ChannelMeta}, "nextOffset": N}. ChannelMeta carries no id field of its own — the id is the map key — so :meth:from_dict takes it as a separate argument rather than reading it out of obj.

from_dict classmethod

from_dict(
    channel_id: str, obj: dict[str, Any]
) -> DomoBuzzChannelMeta

Tolerant of a partial ChannelMeta payload.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

@classmethod
def from_dict(cls, channel_id: str, obj: dict[str, Any]) -> DomoBuzzChannelMeta:
    """Tolerant of a partial ``ChannelMeta`` payload."""
    obj = obj or {}
    return cls(
        id=str(channel_id),
        channel_type=obj.get("channelType"),
        mentioned=obj.get("mentioned"),
        last_viewed_timestamp=obj.get("lastViewedTimestamp"),
        unread_counts=obj.get("unreadCounts"),
        important_counts=obj.get("importantCounts"),
        raw=obj,
    )

DomoBuzzChannels dataclass

DomoBuzzChannels(auth: DomoAuth)

Bases: DomoManager

Collection manager for Buzz channels, messages, topics, and sockets.

authenticate_socket async

authenticate_socket(
    socket_id: str,
    channel_name: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> str | ResponseGetData

Authenticate a private Pusher channel subscription.

POST /buzz/v1/sockets/authenticate is a standard Pusher authorizer: form-urlencoded socket_id/channel_name in, JSON {"auth": "appkey:sig"} out (see routes.buzz.authenticate_socket's docstring for the live finding). Returns the parsed auth string.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def authenticate_socket(
    self,
    socket_id: str,
    channel_name: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> str | ResponseGetData:
    """Authenticate a private Pusher channel subscription.

    ``POST /buzz/v1/sockets/authenticate`` is a standard Pusher
    authorizer: form-urlencoded ``socket_id``/``channel_name`` in,
    JSON ``{"auth": "appkey:sig"}`` out (see
    ``routes.buzz.authenticate_socket``'s docstring for the live
    finding). Returns the parsed ``auth`` string.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await buzz_routes.authenticate_socket(
        auth=self.auth,
        socket_id=socket_id,
        channel_name=channel_name,
        context=context,
    )

    if return_raw:
        return res

    return res.response["auth"]

create_topic async

create_topic(
    title: str,
    description: str | None = None,
    accesses: list[dict[str, Any]] | None = None,
    default_permission: str = "INVITE_OTHERS",
    public_permission: str = "INVITE_OTHERS",
    filters: list[dict[str, Any]] | None = None,
    pin: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoBuzzChannel

Create a Buzz topic.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def create_topic(
    self,
    title: str,
    description: str | None = None,
    accesses: list[dict[str, Any]] | None = None,
    default_permission: str = "INVITE_OTHERS",
    public_permission: str = "INVITE_OTHERS",
    filters: list[dict[str, Any]] | None = None,
    pin: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoBuzzChannel:
    """Create a Buzz topic."""
    return await DomoBuzzChannel.create(
        auth=self.auth,
        title=title,
        description=description,
        accesses=accesses,
        default_permission=default_permission,
        public_permission=public_permission,
        filters=filters,
        pin=pin,
        context=context,
        **context_kwargs,
    )

get async

get(
    offset: int | None = None,
    limit: int | None = None,
    loop_until_end: bool = True,
    maximum: int | None = None,
    debug_loop: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoBuzzChannel] | ResponseGetData

List all Buzz channels visible to the caller (GET /buzz/v1/channels).

Paginated to exhaustion by default so a caller never silently gets a truncated page one — see :func:crew_dcs.routes.buzz.list_channels for the loop_until_end/maximum contract.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def get(
    self,
    offset: int | None = None,
    limit: int | None = None,
    loop_until_end: bool = True,
    maximum: int | None = None,
    debug_loop: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoBuzzChannel] | ResponseGetData:
    """List all Buzz channels visible to the caller (``GET /buzz/v1/channels``).

    Paginated to exhaustion by default so a caller never silently gets a
    truncated page one — see :func:`crew_dcs.routes.buzz.list_channels`
    for the ``loop_until_end``/``maximum`` contract.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await buzz_routes.list_channels(
        auth=self.auth,
        offset=offset,
        limit=limit,
        loop_until_end=loop_until_end,
        maximum=maximum,
        debug_loop=debug_loop,
        context=context,
    )

    if return_raw:
        return res

    channels = (
        res.response.get("channels", []) if isinstance(res.response, dict) else []
    )
    return [DomoBuzzChannel.from_dict(auth=self.auth, obj=ch) for ch in channels]

get_by_id async

get_by_id(
    channel_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoBuzzChannel

Get a single Buzz channel by id.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def get_by_id(
    self,
    channel_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoBuzzChannel:
    """Get a single Buzz channel by id."""
    return await DomoBuzzChannel.get_by_id(
        auth=self.auth,
        channel_id=channel_id,
        context=context,
        **context_kwargs,
    )

get_my_channels async

get_my_channels(
    offset: int | None = None,
    limit: int | None = None,
    loop_until_end: bool = True,
    maximum: int | None = None,
    debug_loop: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoBuzzChannelMeta] | ResponseGetData

List the caller's Buzz channels (GET /buzz/v1/channels/me).

Turns the id-keyed map (response["me"]) into a list of :class:DomoBuzzChannelMeta — one per key, across every page fetched. Paginated to exhaustion by default so a caller never silently gets a truncated page one — see :func:crew_dcs.routes.buzz.list_my_channels for the loop_until_end/maximum contract.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def get_my_channels(
    self,
    offset: int | None = None,
    limit: int | None = None,
    loop_until_end: bool = True,
    maximum: int | None = None,
    debug_loop: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoBuzzChannelMeta] | ResponseGetData:
    """List the caller's Buzz channels (``GET /buzz/v1/channels/me``).

    Turns the id-keyed map (``response["me"]``) into a list of
    :class:`DomoBuzzChannelMeta` — one per key, across every page fetched.
    Paginated to exhaustion by default so a caller never silently gets a
    truncated page one — see
    :func:`crew_dcs.routes.buzz.list_my_channels` for the
    ``loop_until_end``/``maximum`` contract.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await buzz_routes.list_my_channels(
        auth=self.auth,
        offset=offset,
        limit=limit,
        loop_until_end=loop_until_end,
        maximum=maximum,
        debug_loop=debug_loop,
        context=context,
    )

    if return_raw:
        return res

    me_map = res.response.get("me", {}) if isinstance(res.response, dict) else {}
    return [
        DomoBuzzChannelMeta.from_dict(channel_id=channel_id, obj=meta)
        for channel_id, meta in me_map.items()
    ]

get_socket_config async

get_socket_config(
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoBuzzSocketConfig | ResponseGetData

Get the caller's Pusher realtime configuration.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def get_socket_config(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoBuzzSocketConfig | ResponseGetData:
    """Get the caller's Pusher realtime configuration."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await buzz_routes.get_socket_config(auth=self.auth, context=context)

    if return_raw:
        return res

    return DomoBuzzSocketConfig.from_dict(
        res.response if isinstance(res.response, dict) else {}
    )

post_message async

post_message(
    channel: str,
    text: str,
    username: str | None = None,
    icon_url: str | None = None,
    alert: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Post a Slack-shaped incoming-webhook message (POST /buzz/v1/messages).

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

async def post_message(
    self,
    channel: str,
    text: str,
    username: str | None = None,
    icon_url: str | None = None,
    alert: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> ResponseGetData:
    """Post a Slack-shaped incoming-webhook message (``POST /buzz/v1/messages``)."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    return await buzz_routes.post_message(
        auth=self.auth,
        channel=channel,
        text=text,
        username=username,
        icon_url=icon_url,
        alert=alert,
        context=context,
    )

DomoBuzzMessage dataclass

DomoBuzzMessage(
    id: str | None,
    channel_id: str | None = None,
    raw: dict = dict(),
    content: str | dict | None = None,
    guid: str | None = None,
    created: int | None = None,
    created_by_id: str | None = None,
    created_by_type: str | None = None,
)

A Buzz message, as returned in a channel's message history.

Live-verified on datacrew-space 2026-08-28: the wire content field is NOT the plain string sent to create_channel_message — Domo echoes it back wrapped as {"text": "...", "tags": [...]}. Buzz's system messages (bot/integration posts) use a different dict shape — {"fallback": ..., "systemTextKey": ..., "systemTitleKey": ..., "image": ..., ...}, frequently with text present but None — where the human-readable string is fallback, not text.

:attr:content is the resolved text (or, for a dict shape that resolves to neither text nor fallback, the raw content dict itself — never silently None for a dict-shaped payload). The full message object is always on :attr:raw regardless.

from_dict classmethod

from_dict(
    channel_id: str | None, obj: dict[str, Any]
) -> DomoBuzzMessage

Tolerant of a partial message payload.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

@classmethod
def from_dict(cls, channel_id: str | None, obj: dict[str, Any]) -> DomoBuzzMessage:
    """Tolerant of a partial message payload."""
    obj = obj or {}
    content = cls._unwrap_content(obj.get("content"), message_id=obj.get("id"))
    return cls(
        id=obj.get("id"),
        channel_id=obj.get("channelId") or channel_id,
        content=content,
        guid=obj.get("guid"),
        created=obj.get("created"),
        created_by_id=obj.get("createdById"),
        created_by_type=obj.get("createdByType"),
        raw=obj,
    )

DomoBuzzSocketConfig dataclass

DomoBuzzSocketConfig(
    raw: dict = dict(),
    application_key: str | None = None,
    host: str | None = None,
    port: str | None = None,
    ssl_port: str | None = None,
    channel: str | None = None,
    buzz_channel: str | None = None,
)

Pusher configuration from GET /buzz/v1/sockets/me.

from_dict classmethod

from_dict(obj: dict[str, Any]) -> DomoBuzzSocketConfig

Tolerant of a partial socket-config payload.

Source code in src/crew_dcs/classes/DomoBuzz/__init__.py

@classmethod
def from_dict(cls, obj: dict[str, Any]) -> DomoBuzzSocketConfig:
    """Tolerant of a partial socket-config payload."""
    obj = obj or {}
    return cls(
        application_key=obj.get("applicationKey"),
        host=obj.get("host"),
        port=obj.get("port"),
        ssl_port=obj.get("sslPort"),
        channel=obj.get("channel"),
        buzz_channel=obj.get("buzzChannel"),
        raw=obj,
    )