Skip to content

DomoCard

DomoCard

DomoCard Package

This package provides comprehensive card management functionality for Domo instances, including card operations and dataset associations.

Classes:

Name Description
DomoCard_Default

Core card operations and management

DomoCard

Card factory class

DomoCard_DatasetsManager

Manager for datasets associated with a card

DomoCard_OwnerManager

Manager for owners associated with a card

DomoCard_PageManager

Manager for pages associated with a card

DomoCard_DefinitionManager

Manager for kpi definition (columns, formulas, filters)

DomoCardAccessController

Access control controller

Raises:

Type Description
Card_DownloadSourceCodeError

Raised when card source code download fails

BarCardBuilder dataclass

BarCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Vertical bar chart (dimension + measure).

ColumnMapping dataclass

ColumnMapping(
    column: str | None = None,
    formula_id: str | None = None,
    mapping: str | None = None,
    aggregation: str | None = None,
    extras: dict[str, Any] = dict(),
)

A column mapped to a chart axis in a subscription.

Either column (raw dataset field) or formula_id (beast mode reference) will be set, not both.

Attributes:

Name Type Description
column str | None

Raw dataset column name (or None if formula_id)

formula_id str | None

Beast mode calculation ID (or None if raw column)

mapping str | None

Chart axis mapping (e.g., "ITEM", "VALUE", "LABEL")

is_formula property

is_formula: bool

Whether this mapping references a beast mode.

name property

name: str | None

Column name with backticks stripped, or None if formula reference.

to_dict

to_dict() -> dict[str, Any]

Reconstruct the API dict this ColumnMapping was parsed from.

Omits keys whose value is None; preserves any unmodeled keys (alias, format, calendar, ...) captured in extras.

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def to_dict(self) -> dict[str, Any]:
    """Reconstruct the API dict this ColumnMapping was parsed from.

    Omits keys whose value is None; preserves any unmodeled keys
    (``alias``, ``format``, ``calendar``, ...) captured in ``extras``.
    """
    out: dict[str, Any] = {}
    if self.column is not None:
        out["column"] = self.column
    if self.formula_id is not None:
        out["formulaId"] = self.formula_id
    if self.aggregation is not None:
        out["aggregation"] = self.aggregation
    if self.mapping is not None:
        out["mapping"] = self.mapping
    out.update(self.extras)
    return out

ColumnRef dataclass

ColumnRef(column_name: str, column_position: int)

A column reference within a formula's columnPositions.

Attributes:

Name Type Description
column_name str

Column name (may include backticks in raw form)

column_position int

Character position in the formula string

name property

name: str

Column name with backticks stripped.

to_dict

to_dict() -> dict[str, Any]

Reconstruct the API dict this ColumnRef was parsed from.

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
93
94
95
96
97
98
def to_dict(self) -> dict[str, Any]:
    """Reconstruct the API dict this ColumnRef was parsed from."""
    return {
        "columnName": self.column_name,
        "columnPosition": self.column_position,
    }

ColumnSchema dataclass

ColumnSchema(
    id: str = "",
    name: str = "",
    type: str | None = None,
    is_calculation: bool = False,
    is_aggregatable: bool = True,
    source_id: str | None = None,
    hidden: bool = False,
    order: int = 0,
)

A column in the dataset schema from the kpi definition.

Attributes:

Name Type Description
id str

Column ID (usually same as name)

name str

Column display name

type str | None

Data type (e.g., "numeric", "string")

is_calculation bool

Whether this is a beast mode column

is_aggregatable bool

Whether the column can be aggregated

source_id str | None

Source dataset ID

hidden bool

Whether the column is hidden

order int

Column order in the dataset

to_dict

to_dict() -> dict[str, Any]

Reconstruct the API dict for this column schema (typed subset).

NOTE: the parser reads only a subset of the dataset schema fields, so this is intentionally lossy (isEncrypted, isControlled, value, templateId are not modeled and are not emitted).

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
def to_dict(self) -> dict[str, Any]:
    """Reconstruct the API dict for this column schema (typed subset).

    NOTE: the parser reads only a subset of the dataset schema fields, so
    this is intentionally lossy (``isEncrypted``, ``isControlled``,
    ``value``, ``templateId`` are not modeled and are not emitted).
    """
    out: dict[str, Any] = {
        "id": self.id,
        "name": self.name,
        "isCalculation": self.is_calculation,
        "isAggregatable": self.is_aggregatable,
        "hidden": self.hidden,
        "order": self.order,
    }
    if self.type is not None:
        out["type"] = self.type
    if self.source_id is not None:
        out["sourceId"] = self.source_id
    return out

ComboCardBuilder dataclass

ComboCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Combo chart: bar + line on the same card (dimension + measure + series).

Useful for showing amounts as bars and a ratio/trend as a line on the same chart (e.g. Revenue as bars + Margin % as a line).

Set big_number_column to add a summary number above the chart.

DomoCard dataclass

DomoCard(
    auth: DomoAuth,
    id: str,
    raw: dict,
    Lineage: DomoLineage | None = None,
    Definition: DomoCard_DefinitionManager | None = None,
    Datasets: DomoCard_DatasetsManager | None = None,
    Owners: DomoCard_OwnerManager | None = None,
    Pages: DomoCard_PageManager | None = None,
    Access: DomoCardAccessController | None = None,
    BeastModes: DomoCard_BeastModesManager | None = None,
    title: str | None = None,
    description: str | None = None,
    type: str | None = None,
    urn: str | None = None,
    chart_type: str | None = None,
    badge_type: str | None = None,
    dataset_id: str | None = None,
    datastore_id: str | None = None,
    domo_collections: list[Any] = list(),
    domo_source_code: Any = None,
    certification: dict | None = None,
    _init_owners: InitVar[list[Any] | None] = None,
    pages: list[Any] = list(),
)

Bases: DomoCard_Default

DomoCard factory class that uses composition for federated support

create async classmethod

create(
    auth: DomoAuth,
    chart_type: str,
    dataset_id: str,
    title: str,
    *,
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] | None = None,
    order_by: list[Any] | None = None,
    overrides: dict[str, Any] | None = None,
    description: str | None = None,
    definition: dict[str, Any] | None = None,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoCard

Create a new KPI card and return the hydrated :class:DomoCard.

This is the high-level entry point that ties the kpi_builder write side to the create_card route. It builds the card definition from the supplied chart parameters (or accepts a pre-built definition), binds it to dataset_id, and returns a fully-loaded card.

Passing the builder output straight through is deliberately safe: this method sends the inner definition to the route, so callers never have to unwrap the {"definition": ...} envelope themselves.

Parameters:

Name Type Description Default
auth DomoAuth

DomoAuth instance.

required
chart_type str

Friendly chart-type name (e.g. "bar", "single_value"). See kpi_builder.get_registered_card_types().

required
dataset_id str

The dataSourceId to bind the card to.

required
title str

Card title.

required
measure_column str | None

VALUE column (aggregated numeric measure).

None
aggregation str

Aggregation applied to measure_column.

'SUM'
dimension_column str | None

ITEM / primary groupBy column.

None
series_column str | None

Optional SERIES / secondary grouping column.

None
filters list[Any] | None

Optional filters (raw dicts or Filter instances).

None
order_by list[Any] | None

Optional sorts (raw dicts or SortColumn instances).

None
overrides dict[str, Any] | None

Chart overrides (title_x, title_y, footer).

None
description str | None

Optional card description.

None
definition dict[str, Any] | None

Escape hatch — a pre-built inner definition dict (from KpiDefinition.definition or build_kpi_card_definition(...)["definition"]). When given, the chart-parameter args are ignored.

None
return_raw bool

Return the raw create response instead of a DomoCard.

False
context RouteContext | None

Optional RouteContext.

None

Returns:

Type Description
DomoCard

The created card, hydrated via :meth:get_by_id (or the raw

DomoCard

response when return_raw is True).

Raises:

Type Description
CardApiError

If the create call fails.

ValueError

If the create response has no resolvable card id.

Source code in src/crew_dcs/classes/DomoCard/core.py
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@classmethod
async def create(
    cls,
    auth: DomoAuth,
    chart_type: str,
    dataset_id: str,
    title: str,
    *,
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] | None = None,
    order_by: list[Any] | None = None,
    overrides: dict[str, Any] | None = None,
    description: str | None = None,
    definition: dict[str, Any] | None = None,
    return_raw: bool = False,
    context: RouteContext | None = None,
    **context_kwargs,
) -> "DomoCard":
    """Create a new KPI card and return the hydrated :class:`DomoCard`.

    This is the high-level entry point that ties the ``kpi_builder`` write
    side to the ``create_card`` route. It builds the card definition from
    the supplied chart parameters (or accepts a pre-built ``definition``),
    binds it to ``dataset_id``, and returns a fully-loaded card.

    Passing the builder output straight through is deliberately safe: this
    method sends the *inner* definition to the route, so callers never have
    to unwrap the ``{"definition": ...}`` envelope themselves.

    Args:
        auth: DomoAuth instance.
        chart_type: Friendly chart-type name (e.g. ``"bar"``,
            ``"single_value"``). See
            ``kpi_builder.get_registered_card_types()``.
        dataset_id: The dataSourceId to bind the card to.
        title: Card title.
        measure_column: VALUE column (aggregated numeric measure).
        aggregation: Aggregation applied to ``measure_column``.
        dimension_column: ITEM / primary groupBy column.
        series_column: Optional SERIES / secondary grouping column.
        filters: Optional filters (raw dicts or ``Filter`` instances).
        order_by: Optional sorts (raw dicts or ``SortColumn`` instances).
        overrides: Chart overrides (``title_x``, ``title_y``, ``footer``).
        description: Optional card description.
        definition: Escape hatch — a pre-built *inner* definition dict
            (from ``KpiDefinition.definition`` or
            ``build_kpi_card_definition(...)["definition"]``). When given,
            the chart-parameter args are ignored.
        return_raw: Return the raw create response instead of a DomoCard.
        context: Optional RouteContext.

    Returns:
        The created card, hydrated via :meth:`get_by_id` (or the raw
        response when ``return_raw`` is True).

    Raises:
        CardApiError: If the create call fails.
        ValueError: If the create response has no resolvable card id.
    """
    # create + hydrate reuse one pooled session automatically via auth-level
    # connection reuse (RouteContext.reuse_session, default on) — no explicit
    # session handling needed. Lazy import avoids a route<->class cycle.
    from ...routes import card as card_routes
    from .kpi_builder import build_kpi_card_definition

    if definition is None:
        envelope = build_kpi_card_definition(
            chart_type,
            dataset_id=dataset_id,
            title=title,
            measure_column=measure_column,
            aggregation=aggregation,
            measure_formula=measure_formula,
            measure_beastmode=measure_beastmode,
            dimension_column=dimension_column,
            series_column=series_column,
            filters=filters or [],
            order_by=order_by or [],
            overrides=overrides or {},
            description=description,
        )
        definition = envelope["definition"]

    res = await card_routes.create_card(
        auth=auth,
        definition=definition,
        dataset_id=dataset_id,
        context=context,
    )

    if return_raw:
        return res

    body = res.response if isinstance(res.response, dict) else {}
    card_id = body.get("id") or body.get("cardId") or body.get("urn")
    if not card_id:
        raise ValueError(
            f"create_card response did not contain a card id: {res.response!r}"
        )

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

delete async

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

Delete this card.

Source code in src/crew_dcs/classes/DomoCard/core.py
127
128
129
130
131
132
133
134
135
136
137
138
139
async def delete(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> Any:
    """Delete this card."""
    from ...routes import card as card_routes

    context = RouteContext.build_context(context=context, **context_kwargs)
    return await card_routes.delete_card(
        auth=self.auth, card_id=str(self.id), context=context
    )

from_dict classmethod

from_dict(
    auth: DomoAuth,
    obj: dict,
    owners: list[Any] | None = None,
    pages: list[Any] | None = None,
    is_published: bool = False,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    **kwargs
) -> DomoCard

Convert API response dictionary to DomoCard instance.

For federated cards, use composition via card.Federation helper.

Source code in src/crew_dcs/classes/DomoCard/core.py
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
@classmethod
def from_dict(
    cls,
    auth: DomoAuth,
    obj: dict,
    owners: list[Any] | None = None,
    pages: list[Any] | None = None,
    is_published: bool = False,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    **kwargs,
) -> "DomoCard":
    """Convert API response dictionary to DomoCard instance.

    For federated cards, use composition via card.Federation helper.
    """

    # Build the card instance
    card = cls(
        auth=auth,
        id=obj.get("id"),
        raw=obj,
        title=obj.get("title"),
        description=obj.get("description"),
        type=obj.get("type"),
        urn=obj.get("urn"),
        certification=obj.get("certification"),
        chart_type=obj.get("metadata", {}).get("chartType"),
        badge_type=obj.get("metadata", {}).get("chartType"),
        dataset_id=(
            obj.get("datasources", [])[0].get("dataSourceId")
            if obj.get("datasources")
            else None
        ),
        _init_owners=owners or [],
        pages=pages or [],
        datastore_id=obj.get("domoapp", {}).get("id"),
    )

    # Enable federation support if card is federated
    if card.is_federated:
        card.enable_federation_support()

    return card

DomoCardAccessController

DomoCardAccessController(auth: DomoAuth, parent)

Bases: DomoAccessRelationshipController

Access controller for Domo cards.

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
26
27
28
29
def __init__(self, auth: DomoAuth, parent):
    super().__init__(auth=auth, parent_object=parent)
    self.parent = parent
    self.relationships: list[Relationship] = []

add_owners async

add_owners(
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Add owners to the card.

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
async def add_owners(
    self,
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Add owners to the card."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    user_ids = self._collect_ids(domo_users, user_ids)
    group_ids = self._collect_ids(domo_groups, group_ids)

    if not user_ids and not group_ids:
        raise ValueError("Must provide user or group identifiers to add owners")

    existing = await self._get_owner_entries(context=context)
    additions = [{"id": str(uid), "type": "USER"} for uid in user_ids] + [
        {"id": str(gid), "type": "GROUP"} for gid in group_ids
    ]

    combined = {f"{o['type']}:{o['id']}": o for o in existing}
    combined.update({f"{o['type']}:{o['id']}": o for o in additions})

    res = await card_routes.replace_card_owners(
        auth=self.auth,
        card_id=self.parent_id,
        owners=list(combined.values()),
        context=context,
    )

    return res if return_raw else res.is_success

create_relationship async

create_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Create an access relationship for the card.

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def create_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Create an access relationship for the card."""
    if relationship_type == RelationshipType.HAS_ACCESS_OWNER:
        return await self.add_owners(
            user_ids=(
                [target_entity_id]
                if target_entity_type == EntityType.USER
                else None
            ),
            group_ids=(
                [target_entity_id]
                if target_entity_type == EntityType.GROUP
                else None
            ),
            **kwargs,
        )

    return await self.share(
        user_ids=(
            [target_entity_id] if target_entity_type == EntityType.USER else None
        ),
        group_ids=(
            [target_entity_id] if target_entity_type == EntityType.GROUP else None
        ),
        **kwargs,
    )

get_accesslist async

get_accesslist(
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Relationship]

Get access list for the card as relationships.

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
async def get_accesslist(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get access list for the card as relationships."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await card_routes.get_card_access_list(
        auth=self.auth,
        card_id=self.parent_id,
        is_expand_users=True,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    relationships: list[Relationship] = []
    users = res.response.get("users", []) if isinstance(res.response, dict) else []
    groups = (
        res.response.get("groups", []) if isinstance(res.response, dict) else []
    )

    for user in users:
        user_id = user.get("id") or user.get("userId")
        if not user_id:
            continue
        relationships.append(
            Relationship(
                from_entity_id=str(user_id),
                from_entity_type=EntityType.USER,
                to_entity_id=str(self.parent_id),
                to_entity_type=EntityType.CARD,
                relationship_type=self._map_access_level(
                    user.get("accessLevel")
                    or user.get("permission")
                    or user.get("access")
                ),
                metadata={
                    "is_explicit_share": user.get("isExplicitShare", False),
                    **user,
                },
            )
        )

    for group in groups:
        group_id = group.get("id") or group.get("groupId")
        if not group_id:
            continue
        relationships.append(
            Relationship(
                from_entity_id=str(group_id),
                from_entity_type=EntityType.GROUP,
                to_entity_id=str(self.parent_id),
                to_entity_type=EntityType.CARD,
                relationship_type=self._map_access_level(
                    group.get("accessLevel")
                    or group.get("permission")
                    or group.get("access")
                ),
                metadata=group,
            )
        )

    owner_relationships = await self.get_owners(context=context)

    for owner_rel in owner_relationships:
        existing = next(
            (
                r
                for r in relationships
                if r.from_entity_id == owner_rel.from_entity_id
                and r.from_entity_type == owner_rel.from_entity_type
            ),
            None,
        )
        if existing:
            existing.relationship_type = RelationshipType.HAS_ACCESS_OWNER
            existing.metadata["is_owner"] = True
        else:
            relationships.append(owner_rel)

    self.relationships = relationships
    self.invalidate_cache()
    return relationships

get_direct_relationships async

get_direct_relationships() -> list[Relationship]

Get direct relationships for the card.

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
35
36
37
async def get_direct_relationships(self) -> list[Relationship]:
    """Get direct relationships for the card."""
    return await self.get_accesslist()

get_owner_entities async

get_owner_entities(
    is_suppress_errors: bool = True,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Resolve owner entities (users/groups) for this card.

Parameters:

Name Type Description Default
is_suppress_errors bool

When True, suppress DomoError resolution failures.

True
context RouteContext | None

Optional RouteContext for API call configuration

None
**context_kwargs

Additional context parameters

{}

Returns:

Type Description
list[Any]

list of DomoUser/DomoGroup entities

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
async def get_owner_entities(
    self,
    is_suppress_errors: bool = True,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Resolve owner entities (users/groups) for this card.

    Args:
        is_suppress_errors: When True, suppress DomoError resolution failures.
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        list of DomoUser/DomoGroup entities
    """
    from ..DomoGroup.core import DomoGroup
    from ..DomoUser import DomoUser

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

    relationships = await self.get_owners(context=context)

    tasks = []
    for rel in relationships:
        try:
            if rel.from_entity_type == EntityType.USER:
                tasks.append(
                    DomoUser.get_by_id(
                        auth=self.auth,
                        user_id=rel.from_entity_id,
                        context=context,
                    )
                )
            elif rel.from_entity_type == EntityType.GROUP:
                tasks.append(
                    DomoGroup.get_by_id(
                        auth=self.auth,
                        group_id=rel.from_entity_id,
                        context=context,
                    )
                )
        except DomoError:
            if not is_suppress_errors:
                raise

    if not tasks:
        return []

    return await dmce.gather_with_concurrency(n=60, *tasks)  # noqa: B026

get_owners async

get_owners(
    *, context: RouteContext | None = None, **context_kwargs
) -> list[Relationship]

Get owner relationships for the card.

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
async def get_owners(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get owner relationships for the card."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await card_routes.get_card_metadata(
        auth=self.auth,
        card_id=self.parent_id,
        optional_parts="owners",
        context=context,
    )

    owners = (
        res.response.get("owners", []) if isinstance(res.response, dict) else []
    )

    relationships: list[Relationship] = []
    for owner in owners:
        owner_id = owner.get("id")
        owner_type = owner.get("type", "USER")
        if not owner_id:
            continue
        relationships.append(
            Relationship(
                from_entity_id=str(owner_id),
                from_entity_type=(
                    EntityType.GROUP if owner_type == "GROUP" else EntityType.USER
                ),
                to_entity_id=str(self.parent_id),
                to_entity_type=EntityType.CARD,
                relationship_type=RelationshipType.HAS_ACCESS_OWNER,
                metadata={"is_owner": True, **owner},
            )
        )

    return relationships

remove_owners async

remove_owners(
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Remove owners from the card.

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
async def remove_owners(
    self,
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Remove owners from the card."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    user_ids = set(self._collect_ids(domo_users, user_ids))
    group_ids = set(self._collect_ids(domo_groups, group_ids))

    existing = await self._get_owner_entries(context=context)
    remaining = [
        owner
        for owner in existing
        if not (
            (owner.get("type") == "USER" and owner.get("id") in user_ids)
            or (owner.get("type") == "GROUP" and owner.get("id") in group_ids)
        )
    ]

    if not remaining:
        raise ValueError("Cannot remove all owners; owners would be empty")

    res = await card_routes.replace_card_owners(
        auth=self.auth,
        card_id=self.parent_id,
        owners=remaining,
        context=context,
    )

    return res if return_raw else res.is_success

remove_relationship async

remove_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Remove an access relationship for the card.

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
async def remove_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Remove an access relationship for the card."""
    if relationship_type == RelationshipType.HAS_ACCESS_OWNER:
        return await self.remove_owners(
            user_ids=(
                [target_entity_id]
                if target_entity_type == EntityType.USER
                else None
            ),
            group_ids=(
                [target_entity_id]
                if target_entity_type == EntityType.GROUP
                else None
            ),
            **kwargs,
        )

    raise NotImplementedError(
        "Removing card share relationships is not supported by current routes."
    )

share async

share(
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    message: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Share the card with users or groups.

Source code in src/crew_dcs/classes/DomoCard/access_controller.py
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
async def share(
    self,
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    message: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Share the card with users or groups."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    user_ids = self._collect_ids(domo_users, user_ids)
    group_ids = self._collect_ids(domo_groups, group_ids)

    if not user_ids and not group_ids:
        raise ValueError("Must provide user or group identifiers to share card")

    res = await share_resource(
        auth=self.auth,
        resource_ids=self.parent_id,
        resource_type=ShareResource_Enum.CARD,
        group_ids=group_ids,
        user_ids=user_ids,
        message=message,
        return_raw=return_raw,
        context=context,
    )

    return res if return_raw else res.is_success

DomoCard_BeastModesManager dataclass

DomoCard_BeastModesManager(
    parent: DomoEntity,
    beast_modes: list[DomoBeastMode] = list(),
)

Bases: DomoSubEntity

Manager for beast modes associated with a DomoCard.

Provides typed access to all beast modes linked to a card.

Usage

bms = await card.BeastModes.get() card.BeastModes.beast_modes # list[DomoBeastMode]

get async

get(
    *, context: RouteContext | None = None, **context_kwargs
) -> list[DomoBeastMode]

Fetch beast modes linked to this card.

Delegates to DomoBeastModes.get_by_card and caches the result.

Parameters:

Name Type Description Default
context RouteContext | None

Optional RouteContext for request configuration

None
**context_kwargs

Additional context parameters

{}

Returns:

Type Description
list[DomoBeastMode]

List of DomoBeastMode instances linked to this card

Source code in src/crew_dcs/classes/DomoCard/manager_beastmodes.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
async def get(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoBeastMode]:  # noqa: F821
    """Fetch beast modes linked to this card.

    Delegates to DomoBeastModes.get_by_card and caches the result.

    Args:
        context: Optional RouteContext for request configuration
        **context_kwargs: Additional context parameters

    Returns:
        List of DomoBeastMode instances linked to this card
    """
    from ..DomoBeastMode.manager import DomoBeastModes

    self.beast_modes = await DomoBeastModes(auth=self.parent.auth).get_by_card(
        card_id=self.parent.id,
        context=context,
        **context_kwargs,
    )
    return self.beast_modes

DomoCard_DatasetsManager dataclass

DomoCard_DatasetsManager(
    auth: DomoAuth,
    parent: DomoCard_Default = None,
    datasets: list[Any] = list(),
)

Bases: DomoManager

Manager for datasets associated with a DomoCard.

Provides access to all datasets used by a card through its datasources. Inherits from DomoManager to follow standard entity manager patterns.

get async

get(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    parent_auth_retrieval_fn: (
        Callable[[str], Any] | None
    ) = None,
    **context_kwargs
) -> list[Any]

Get all datasets associated with this card.

This retrieves datasets from the card's datasources and returns DomoDataset instances for each one.

Parameters:

Name Type Description Default
debug_api bool

Enable API debugging

False
session AsyncClient | None

Optional httpx session for request reuse

None

Returns:

Type Description
list[Any]

list[DomoDataset]: List of dataset objects associated with the card

Source code in src/crew_dcs/classes/DomoCard/manager_dataset.py
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def get(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    parent_auth_retrieval_fn: Callable[[str], Any] | None = None,
    **context_kwargs,
) -> list[Any]:  # Returns list[DomoDataset]
    """Get all datasets associated with this card.

    This retrieves datasets from the card's datasources and returns
    DomoDataset instances for each one.

    Args:
        debug_api: Enable API debugging
        session: Optional httpx session for request reuse

    Returns:
        list[DomoDataset]: List of dataset objects associated with the card
    """
    from ..DomoDataset import DomoDataset

    context = RouteContext.build_context(
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=1,
        **context_kwargs,
    )
    # Get datasources from card metadata if not already loaded
    if not self.parent.raw.get("datasources"):
        # Reload card with datasources
        res = await card_routes.get_card_metadata(
            auth=self.auth,
            card_id=self.parent.id,
            optional_parts="datasources",
            context=context,
        )
        self.parent.raw = res.response

    datasources = self.parent.raw.get("datasources", [])

    if not datasources:
        return []

    # Get dataset IDs from datasources
    dataset_ids = [
        ds.get("dataSourceId") for ds in datasources if ds.get("dataSourceId")
    ]

    if not dataset_ids:
        return []

    # Fetch all datasets concurrently
    # Auto-enable publish check and lineage tracing when parent_auth_retrieval_fn is provided
    datasets = await dmce.gather_with_concurrency(
        *[
            DomoDataset.get_by_id(
                auth=self.auth,
                dataset_id=dataset_id,
                check_if_published=bool(parent_auth_retrieval_fn),
                parent_auth_retrieval_fn=parent_auth_retrieval_fn,
                context=context,
            )
            for dataset_id in dataset_ids
        ],
        n=60,
    )

    self.datasets = datasets
    return datasets

DomoCard_Default dataclass

DomoCard_Default(
    auth: DomoAuth,
    id: str,
    raw: dict,
    Lineage: DomoLineage | None = None,
    Definition: DomoCard_DefinitionManager | None = None,
    Datasets: DomoCard_DatasetsManager | None = None,
    Owners: DomoCard_OwnerManager | None = None,
    Pages: DomoCard_PageManager | None = None,
    Access: DomoCardAccessController | None = None,
    BeastModes: DomoCard_BeastModesManager | None = None,
    title: str | None = None,
    description: str | None = None,
    type: str | None = None,
    urn: str | None = None,
    chart_type: str | None = None,
    badge_type: str | None = None,
    dataset_id: str | None = None,
    datastore_id: str | None = None,
    domo_collections: list[Any] = list(),
    domo_source_code: Any = None,
    certification: dict | None = None,
    _init_owners: InitVar[list[Any] | None] = None,
    pages: list[Any] = list(),
)

Bases: DomoEntity_w_Lineage

Base DomoCard implementation with core functionality

datasets property

datasets: list[Any]

Legacy property access - prefer using Datasets.get() for async operations

entity_name property

entity_name: str

Get the display name for this card.

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

Returns:

Type Description
str

Card title, or card ID as fallback

is_federated property

is_federated: bool

Check if this card is federated

name property

name: str

Get the display name for this card.

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

Returns:

Type Description
str

Card title, or "Untitled Card {id}" as fallback

owners property

owners: list[Any]

Return owners from the Owners manager when available.

add_owners async

add_owners(
    owners: list[Any],
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Add owners (users or groups) to this card.

Source code in src/crew_dcs/classes/DomoCard/card_default.py
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
async def add_owners(
    self,
    owners: list[Any],
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Add owners (users or groups) to this card."""
    if not self.Owners:
        self.Owners = DomoCard_OwnerManager.from_parent(auth=self.auth, parent=self)

    return await self.Owners.add_owners(
        owners=owners,
        debug_api=debug_api,
        session=session,
        context=context,
        **context_kwargs,
    )

from_dict classmethod

from_dict(
    auth: DomoAuth,
    obj: dict[str, Any],
    owners: list[Any] | None = None,
    pages: list[Any] | None = None,
)

Synchronous factory method for dict → Card construction.

Build a DomoCard_Default instance from a metadata dictionary and owners list.

This method does not invoke any other class methods or perform additional API calls. All required data must be provided in the arguments.

Parameters:

Name Type Description Default
auth DomoAuth

DomoAuth instance for authentication

required
obj dict[str, Any]

Card metadata dictionary (from API)

required
owners list[Any] | None

List of owner entities (DomoUser/DomoGroup), optional

None

Returns:

Type Description

DomoCard_Default instance

Source code in src/crew_dcs/classes/DomoCard/card_default.py
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
@classmethod
def from_dict(
    cls,
    auth: DomoAuth,
    obj: dict[str, Any],
    owners: list[Any] | None = None,
    pages: list[Any] | None = None,
):
    """Synchronous factory method for dict → Card construction.

    Build a DomoCard_Default instance from a metadata dictionary and owners list.

    This method does not invoke any other class methods or perform additional API calls.
    All required data must be provided in the arguments.

    Args:
        auth: DomoAuth instance for authentication
        obj: Card metadata dictionary (from API)
        owners: List of owner entities (DomoUser/DomoGroup), optional

    Returns:
        DomoCard_Default instance
    """
    owners = owners or []
    pages = pages or []

    card = cls(
        auth=auth,
        id=obj.get("id"),
        raw=obj,
        title=obj.get("title"),
        description=obj.get("description"),
        type=obj.get("type"),
        urn=obj.get("urn"),
        certification=obj.get("certification"),
        chart_type=obj.get("metadata", {}).get("chartType"),
        badge_type=obj.get("metadata", {}).get("chartType"),  # same as chartType
        dataset_id=(
            obj.get("datasources", [])[0].get("dataSourceId")
            if obj.get("datasources")
            else None
        ),
        _init_owners=owners,
        pages=pages,
        datastore_id=obj.get("domoapp", {}).get("id"),
    )

    return card  # noqa: RET504

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    card_id: str,
    optional_parts: str = "metadata,certification,datasources,drillPath,owners,properties,domoapp",
    check_if_published: bool | None = None,
    resolve_pages: bool | None = None,
    resolve_datasources: bool | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    max_subscriptions_to_check: int | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    is_suppress_errors: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Retrieve a DomoCard by ID, including publication/certification status.

Parameters:

Name Type Description Default
auth DomoAuth

DomoAuth instance

required
card_id str

Card ID to retrieve

required
optional_parts str

Comma-separated metadata parts to include (owners, certification, etc.)

'metadata,certification,datasources,drillPath,owners,properties,domoapp'
check_if_published bool | None

When True, checks if card is published (federated + subscriptions). If None and either parent_auth or parent_auth_retrieval_fn is provided, defaults to True.

None
resolve_pages bool | None

When True, resolve page entities from metadata and link in lineage. If None, preserves current behavior (resolve pages if present).

None
resolve_datasources bool | None

When True, resolve dataset entities from datasources and link in lineage. If None, defaults to False (no dataset resolution).

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
max_subscriptions_to_check int | None

Optional limit when scanning subscriptions

None
debug_api bool

Enable API debug logging

False
session AsyncClient | None

Optional httpx session

None
return_raw bool

If True, return raw API response

False
is_suppress_errors bool

If True, suppress errors during owner resolution (handled via Access)

False
context RouteContext | None

Optional RouteContext for API call configuration

None
**context_kwargs

Additional context parameters

{}

Returns:

Type Description

DomoCard_Default instance (or raw response if return_raw)

Processing steps
  1. Fetch card metadata from API (includes owners, certification, datasources, etc.)
  2. If return_raw, return API response directly
  3. If card is federated and check_if_published is True, run additional check for publication status
  4. Build DomoCard_Default instance
Source code in src/crew_dcs/classes/DomoCard/card_default.py
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
@classmethod
@log_call(
    level_name="entity",
    config=LogDecoratorConfig(result_processor=DomoEntityObjectProcessor()),
)
async def get_by_id(  # noqa: C901
    cls,
    auth: DomoAuth,
    card_id: str,
    optional_parts: str = "metadata,certification,datasources,drillPath,owners,properties,domoapp",
    check_if_published: bool | None = None,
    resolve_pages: bool | None = None,
    resolve_datasources: bool | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    max_subscriptions_to_check: int | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    is_suppress_errors: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """
    Retrieve a DomoCard by ID, including publication/certification status.

    Args:
        auth: DomoAuth instance
        card_id: Card ID to retrieve
        optional_parts: Comma-separated metadata parts to include (owners, certification, etc.)
        check_if_published: When True, checks if card is published (federated + subscriptions).
            If None and either parent_auth or parent_auth_retrieval_fn is provided, defaults to True.
        resolve_pages: When True, resolve page entities from metadata and link in lineage.
            If None, preserves current behavior (resolve pages if present).
        resolve_datasources: When True, resolve dataset entities from datasources and link in lineage.
            If None, defaults to False (no dataset resolution).
        parent_auth_retrieval_fn: Callable returning publisher auth when given publisher domain
        parent_auth: Pre-existing publisher auth (alternative to parent_auth_retrieval_fn)
        max_subscriptions_to_check: Optional limit when scanning subscriptions
        debug_api: Enable API debug logging
        session: Optional httpx session
        return_raw: If True, return raw API response
        is_suppress_errors: If True, suppress errors during owner resolution (handled via Access)
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        DomoCard_Default instance (or raw response if return_raw)

    Processing steps:
        1. Fetch card metadata from API (includes owners, certification, datasources, etc.)
        2. If return_raw, return API response directly
        3. If card is federated and check_if_published is True, run additional check for publication status
        4. Build DomoCard_Default 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

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

    resolve_pages = True if resolve_pages is None else resolve_pages
    resolve_datasources = (
        False if resolve_datasources is None else resolve_datasources
    )

    res = await card_routes.get_card_metadata(
        auth=auth,
        card_id=card_id,
        optional_parts=optional_parts,
        context=context,
    )

    if return_raw:
        return res

    pages: list[Any] = []

    # Check if published (if federated and check enabled)
    is_published = False
    subscription = None
    if (
        check_if_published
        and cls._is_federated(res.response)
        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(card_id),
            entity_type="CARD",
        )
        is_published = await probe.check_if_published(
            retrieve_parent_auth_fn=effective_retrieval_fn,
            entity_type="CARD",
            session=session,
            debug_api=debug_api,
            max_subscriptions_to_check=max_subscriptions_to_check,
            context=context,
        )
        if is_published:
            subscription = probe.subscription

    domo_card = cls.from_dict(
        auth=auth,
        obj=res.response,
        owners=[],
        pages=pages,
    )

    if resolve_pages and domo_card.Pages:
        pages = await domo_card.Pages.get(context=context)
        domo_card.pages = pages

    datasets: list[Any] = []
    if resolve_datasources and domo_card.Datasets:
        datasets = await domo_card.Datasets.get(
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            context=context,
        )

    if domo_card.Lineage and (resolve_pages or resolve_datasources):
        from ..DomoDataset.lineage import DomoLineageLink_Dataset
        from ..DomoPage.lineage import DomoLineageLink_Page
        from .lineage import DomoLineageLink_Card

        card_link = DomoLineageLink_Card(
            auth=domo_card.auth,
            id=str(domo_card.id),
            entity=domo_card,
            _type=None,
            dependents=[],
            dependencies=[],
        )

        dataset_links: list[Any] = []
        for dataset in datasets:
            dataset_link = DomoLineageLink_Dataset(
                auth=dataset.auth,
                id=str(dataset.id),
                entity=dataset,
                _type=None,
                dependencies=[],
                dependents=[card_link],
            )
            dataset_links.append(dataset_link)
            card_link.dependencies.append(dataset_link)

        page_links: list[Any] = []
        for page in pages:
            page_link = DomoLineageLink_Page(
                auth=page.auth,
                id=str(page.id),
                entity=page,
                _type=None,
                dependencies=[card_link],
                dependents=[],
            )
            page_links.append(page_link)
            card_link.dependents.append(page_link)

        domo_card.Lineage.lineage = [card_link, *dataset_links, *page_links]
        domo_card.Lineage.immediate_dependencies = list(card_link.dependencies)
        domo_card.Lineage.immediate_dependents = list(card_link.dependents)

    if is_published and subscription:
        helper = domo_card.enable_federation_support()
        helper.hydrate_from_existing(
            subscription=subscription,
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            parent_auth=parent_auth,
            content_type="CARD",
            entity_id=str(card_id),
        )

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

    return domo_card

get_owners async

get_owners(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Refresh owners for this card.

Source code in src/crew_dcs/classes/DomoCard/card_default.py
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
async def get_owners(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Refresh owners for this card."""
    if not self.Owners:
        self.Owners = DomoCard_OwnerManager.from_parent(auth=self.auth, parent=self)

    return await self.Owners.get(
        debug_api=debug_api,
        session=session,
        context=context,
        **context_kwargs,
    )

get_pages async

get_pages(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Refresh pages this card appears on.

Source code in src/crew_dcs/classes/DomoCard/card_default.py
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
async def get_pages(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Refresh pages this card appears on."""
    if not self.Pages:
        self.Pages = DomoCard_PageManager.from_parent(auth=self.auth, parent=self)

    return await self.Pages.get(
        debug_api=debug_api,
        session=session,
        context=context,
        **context_kwargs,
    )

hydrate_card_metadata async classmethod

hydrate_card_metadata(
    cards: list[DomoCard_Default],
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> None

Batch-hydrate card metadata (chartType/badge_type) from the API.

The lineage API does not include card metadata. This method fetches metadata for all cards in a single batch request and updates their badge_type and chart_type fields in-place.

Parameters:

Name Type Description Default
cards list[DomoCard_Default]

List of DomoCard instances to hydrate

required
auth DomoAuth

Authentication object

required
context RouteContext | None

Optional RouteContext

None
**context_kwargs

Additional context parameters

{}
Source code in src/crew_dcs/classes/DomoCard/card_default.py
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
@classmethod
async def hydrate_card_metadata(
    cls,
    cards: list[DomoCard_Default],
    auth: DomoAuth,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> None:
    """Batch-hydrate card metadata (chartType/badge_type) from the API.

    The lineage API does not include card metadata. This method fetches
    metadata for all cards in a single batch request and updates their
    badge_type and chart_type fields in-place.

    Args:
        cards: List of DomoCard instances to hydrate
        auth: Authentication object
        context: Optional RouteContext
        **context_kwargs: Additional context parameters
    """
    if not cards:
        return

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

    # Filter to cards that don't already have badge_type
    cards_needing_hydration = [c for c in cards if not c.badge_type]
    if not cards_needing_hydration:
        return

    card_ids = [str(c.id) for c in cards_needing_hydration]

    res = await card_routes.get_cards_by_ids(
        card_ids=card_ids,
        auth=auth,
        context=context,
    )

    if not res.is_success:
        return

    # Build lookup by ID
    card_map = {str(c["id"]): c for c in res.response}

    for card in cards_needing_hydration:
        card_data = card_map.get(str(card.id))
        if not card_data:
            continue

        metadata = card_data.get("metadata", {})
        chart_type = metadata.get("chartType")
        if chart_type:
            card.badge_type = chart_type
            card.chart_type = chart_type

        # Also update raw if present
        if card.raw:
            card.raw.setdefault("metadata", {}).update(metadata)

remove_owners async

remove_owners(
    owners: list[Any],
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Remove owners (users or groups) from this card.

Source code in src/crew_dcs/classes/DomoCard/card_default.py
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
async def remove_owners(
    self,
    owners: list[Any],
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Remove owners (users or groups) from this card."""
    if not self.Owners:
        self.Owners = DomoCard_OwnerManager.from_parent(auth=self.auth, parent=self)

    return await self.Owners.remove_owners(
        owners=owners,
        debug_api=debug_api,
        session=session,
        context=context,
        **context_kwargs,
    )

to_dict

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

Serialize card for export/debugging with safe owner serialization.

Source code in src/crew_dcs/classes/DomoCard/card_default.py
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def to_dict(
    self, override_fn: Callable | None = None, return_snake_case: bool = False
) -> dict:
    """Serialize card for export/debugging with safe owner serialization."""
    if override_fn:
        return override_fn(self)

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

    owners_key = "owners" if return_snake_case else "Owners"
    pages_key = "pages" if return_snake_case else "Pages"

    def _serialize_owner(owner: Any) -> Any:
        if owner is None:
            return None
        if isinstance(owner, dict):
            return owner

        owner_id = getattr(owner, "id", None)
        owner_type = getattr(owner, "entity_type", None) or owner.__class__.__name__
        display_name = getattr(owner, "display_name", None) or getattr(
            owner, "name", None
        )

        if return_snake_case:
            owner_dict = {"id": owner_id, "entity_type": owner_type}
            if display_name:
                owner_dict["display_name"] = display_name
        else:
            owner_dict = {"id": owner_id, "entityType": owner_type}
            if display_name:
                owner_dict["displayName"] = display_name

        return owner_dict

    if self.owners:
        result[owners_key] = [
            o for o in (_serialize_owner(o) for o in self.owners) if o is not None
        ]

    def _serialize_page(page: Any) -> Any:
        return self._serialize_related_entity(
            page,
            return_snake_case=return_snake_case,
            include_entity_type=False,
        )

    if self.pages:
        result[pages_key] = [
            p for p in (_serialize_page(p) for p in self.pages) if p is not None
        ]

    return result

DomoCard_DefinitionManager dataclass

DomoCard_DefinitionManager(
    parent: DomoCard_Default = None,
    kpi: KpiDefinition | None = None,
)

Bases: DomoSubEntity

Manager for a card's kpi definition.

Provides typed access to: - KpiDefinition with subscriptions, formulas, and column schema - Convenience properties: used_columns, available_columns, unused_columns

Usage

await card.Definition.get() card.Definition.kpi # KpiDefinition card.Definition.used_columns # set[str] card.Definition.available_columns # list[str] card.Definition.kpi.main.columns # list[ColumnMapping] card.Definition.kpi.formula_by_id # dict[str, Formula]

With dataset beast mode resolution

await card.Definition.get(is_resolve_dataset_beastmodes=True)

Now formula_by_id includes dataset-level beast modes

and filter/group_by columns resolve calculation_xxx IDs

available_columns property

available_columns: list[str]

All available columns from the dataset schema. Empty if not loaded.

unused_columns property

unused_columns: set[str]

Dataset columns NOT used by this card. Empty if not loaded.

used_columns property

used_columns: set[str]

Columns used by this card. Empty if definition not loaded.

get async

get(
    is_resolve_dataset_beastmodes: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> KpiDefinition

Fetch the kpi definition for this card.

Populates self.kpi with a typed KpiDefinition.

Parameters:

Name Type Description Default
is_resolve_dataset_beastmodes bool

If True, also fetch the card's datasets and resolve any calculation_xxx formula IDs that reference dataset-level beast modes. This ensures formula_by_id and used_columns correctly resolve beast modes saved to the dataset (not on the card itself).

False
context RouteContext | None

Optional RouteContext for request configuration

None

Returns:

Type Description
KpiDefinition

KpiDefinition instance with typed access to all definition data.

Source code in src/crew_dcs/classes/DomoCard/manager_definition.py
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def get(
    self,
    is_resolve_dataset_beastmodes: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> KpiDefinition:
    """Fetch the kpi definition for this card.

    Populates self.kpi with a typed KpiDefinition.

    Args:
        is_resolve_dataset_beastmodes: If True, also fetch the card's
            datasets and resolve any ``calculation_xxx`` formula IDs
            that reference dataset-level beast modes. This ensures
            ``formula_by_id`` and ``used_columns`` correctly resolve
            beast modes saved to the dataset (not on the card itself).
        context: Optional RouteContext for request configuration

    Returns:
        KpiDefinition instance with typed access to all definition data.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await card_routes.get_kpi_definition(
        auth=self.auth,
        card_id=self.parent.id,
        context=context,
    )

    if res.is_success:
        self.kpi = KpiDefinition.from_dict(res.response)

    if is_resolve_dataset_beastmodes and self.kpi:
        await self._resolve_dataset_beastmodes(context=context)

    return self.kpi

DomoCard_OwnerManager dataclass

DomoCard_OwnerManager(
    auth: DomoAuth | None = None,
    parent: DomoCard_Default | None = None,
    owners: list[Any] = list(),
)

Bases: DomoRelationshipController

Manager for owners associated with a DomoCard.

add_owners async

add_owners(
    owners: list[Any],
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Add owners (users or groups) to this card.

Source code in src/crew_dcs/classes/DomoCard/manager_owner.py
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
async def add_owners(
    self,
    owners: list[Any],
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Add owners (users or groups) to this card."""
    if not owners:
        raise ValueError("owners must be a non-empty list")

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

    existing_owners = await self.get(context=context)
    existing = [self._normalize_owner(owner) for owner in existing_owners]
    existing = [owner for owner in existing if owner]

    additions = [self._normalize_owner(owner) for owner in owners]
    additions = [owner for owner in additions if owner]

    if not additions:
        raise ValueError("owners did not contain valid user or group entries")

    combined = {f"{o['type']}:{o['id']}": o for o in existing}
    combined.update({f"{o['type']}:{o['id']}": o for o in additions})

    await card_routes.replace_card_owners(
        auth=self.auth,
        card_id=self.parent.id,
        owners=list(combined.values()),
        context=context,
    )

    return await self.get(context=context)

create_relationship async

create_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Create an owner relationship for this card.

Source code in src/crew_dcs/classes/DomoCard/manager_owner.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def create_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Create an owner relationship for this card."""
    if relationship_type != RelationshipType.HAS_ACCESS_OWNER:
        raise ValueError(
            "Only owner relationships are supported by DomoCard_OwnerManager"
        )

    if target_entity_type == EntityType.USER:
        await self.add_owners([{"id": target_entity_id, "type": "USER"}], **kwargs)
        return True

    if target_entity_type == EntityType.GROUP:
        await self.add_owners([{"id": target_entity_id, "type": "GROUP"}], **kwargs)
        return True

    raise ValueError(f"Unsupported owner entity type: {target_entity_type}")

get async

get(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Get owners for this card.

Source code in src/crew_dcs/classes/DomoCard/manager_owner.py
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
async def get(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:  # Returns list[DomoUser|DomoGroup]
    """Get owners for this card."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=1,
        **context_kwargs,
    )

    if not self.parent.Access:
        from .access_controller import DomoCardAccessController

        self.parent.Access = DomoCardAccessController.from_parent(
            parent=self.parent
        )

    owners = await self.parent.Access.get_owner_entities(
        is_suppress_errors=True,
        context=context,
    )

    self.owners = owners
    self.parent.owners = owners

    return owners

get_direct_relationships async

get_direct_relationships() -> list[Relationship]

Get owner relationships for this card.

Source code in src/crew_dcs/classes/DomoCard/manager_owner.py
38
39
40
41
42
43
44
45
46
47
async def get_direct_relationships(self) -> list[Relationship]:
    """Get owner relationships for this card."""
    if not self.parent.Access:
        from .access_controller import DomoCardAccessController

        self.parent.Access = DomoCardAccessController.from_parent(
            parent=self.parent
        )

    return await self.parent.Access.get_owners()

remove_owners async

remove_owners(
    owners: list[Any],
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Remove owners (users or groups) from this card.

Source code in src/crew_dcs/classes/DomoCard/manager_owner.py
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
async def remove_owners(
    self,
    owners: list[Any],
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Remove owners (users or groups) from this card."""
    if not owners:
        raise ValueError("owners must be a non-empty list")

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

    existing_owners = await self.get(context=context)
    existing = [self._normalize_owner(owner) for owner in existing_owners]
    existing = [owner for owner in existing if owner]

    removals = [self._normalize_owner(owner) for owner in owners]
    removals = [owner for owner in removals if owner]

    if not removals:
        raise ValueError("owners did not contain valid user or group entries")

    remove_keys = {f"{o['type']}:{o['id']}" for o in removals}
    updated = [
        owner
        for owner in existing
        if f"{owner['type']}:{owner['id']}" not in remove_keys
    ]

    if not updated:
        raise ValueError("Cannot remove all owners; owners would be empty")

    await card_routes.replace_card_owners(
        auth=self.auth,
        card_id=self.parent.id,
        owners=updated,
        context=context,
    )

    return await self.get(context=context)

remove_relationship async

remove_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Remove an owner relationship for this card.

Source code in src/crew_dcs/classes/DomoCard/manager_owner.py
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
async def remove_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Remove an owner relationship for this card."""
    if relationship_type != RelationshipType.HAS_ACCESS_OWNER:
        raise ValueError(
            "Only owner relationships are supported by DomoCard_OwnerManager"
        )

    if target_entity_type == EntityType.USER:
        await self.remove_owners(
            [{"id": target_entity_id, "type": "USER"}], **kwargs
        )
        return True

    if target_entity_type == EntityType.GROUP:
        await self.remove_owners(
            [{"id": target_entity_id, "type": "GROUP"}], **kwargs
        )
        return True

    raise ValueError(f"Unsupported owner entity type: {target_entity_type}")

DomoCard_PageManager dataclass

DomoCard_PageManager(
    auth: DomoAuth | None = None,
    parent: DomoCard_Default | None = None,
    pages: list[Any] = list(),
)

Bases: DomoRelationshipController

Manager for pages associated with a DomoCard.

add_pages async

add_pages(
    page_ids: list[str | int],
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Add this card to the specified pages.

Source code in src/crew_dcs/classes/DomoCard/manager_page.py
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
async def add_pages(
    self,
    page_ids: list[str | int],
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Add this card to the specified pages."""
    if not page_ids:
        raise ValueError("page_ids must be a non-empty list")

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

    existing_pages = await self.get(context=context)
    existing_ids = {str(getattr(pg, "id", pg)) for pg in existing_pages}
    updated_ids = sorted({*existing_ids, *{str(pid) for pid in page_ids}})

    await card_routes.replace_card_pages(
        auth=self.auth,
        card_id=self.parent.id,
        page_ids=updated_ids,
        context=context,
    )

    return await self.get(context=context)

create_relationship async

create_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Create a card -> page relationship by adding the card to a page.

Source code in src/crew_dcs/classes/DomoCard/manager_page.py
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def create_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Create a card -> page relationship by adding the card to a page."""
    if relationship_type != RelationshipType.CONTAINS:
        raise ValueError(
            "Only CONTAINS relationships are supported by DomoCard_PageManager"
        )

    if target_entity_type != EntityType.PAGE:
        raise ValueError(
            "Only PAGE relationships are supported by DomoCard_PageManager"
        )

    await self.add_pages([target_entity_id], **kwargs)
    return True

get async

get(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Get pages that include this card.

Source code in src/crew_dcs/classes/DomoCard/manager_page.py
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
async def get(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:  # Returns list[DomoPage]
    """Get pages that include this card."""
    from ..DomoPage import DomoPage

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

    res = await card_routes.get_card_metadata(
        auth=self.auth,
        card_id=self.parent.id,
        optional_parts="pages",
        context=context,
    )

    page_items = res.response.get("pages", [])
    if not page_items:
        self.pages = []
        self.parent.pages = []
        return []

    page_ids: list[str] = []
    for page in page_items:
        page_id = None
        if isinstance(page, dict):
            page_id = page.get("id") or page.get("pageId") or page.get("page_id")
        else:
            page_id = getattr(page, "id", None) or getattr(page, "pageId", None)

        if page_id is not None:
            page_ids.append(str(page_id))

    if not page_ids:
        self.pages = []
        self.parent.pages = []
        return []

    pages = await dmce.gather_with_concurrency(
        n=60,
        *[  # noqa: B026
            DomoPage.get_by_id(auth=self.auth, page_id=page_id, context=context)
            for page_id in page_ids
        ],
    )

    self.pages = pages
    self.parent.pages = pages

    return pages

get_direct_relationships async

get_direct_relationships() -> list[Relationship]

Get relationships between this card and its pages.

Source code in src/crew_dcs/classes/DomoCard/manager_page.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
async def get_direct_relationships(self) -> list[Relationship]:
    """Get relationships between this card and its pages."""
    pages = await self.get()
    return [
        Relationship(
            from_entity_id=str(self.parent.id),
            from_entity_type=EntityType.CARD,
            to_entity_id=str(getattr(page, "id", page)),
            to_entity_type=EntityType.PAGE,
            relationship_type=RelationshipType.CONTAINS,
            metadata={"card_id": str(self.parent.id)},
        )
        for page in pages
    ]

remove_pages async

remove_pages(
    page_ids: list[str | int],
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Remove this card from the specified pages.

Source code in src/crew_dcs/classes/DomoCard/manager_page.py
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
async def remove_pages(
    self,
    page_ids: list[str | int],
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Remove this card from the specified pages."""
    if not page_ids:
        raise ValueError("page_ids must be a non-empty list")

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

    existing_pages = await self.get(context=context)
    existing_ids = {str(getattr(pg, "id", pg)) for pg in existing_pages}
    updated_ids = sorted(existing_ids.difference({str(pid) for pid in page_ids}))

    if not updated_ids:
        raise ValueError("Cannot remove all pages; page_ids would be empty")

    await card_routes.replace_card_pages(
        auth=self.auth,
        card_id=self.parent.id,
        page_ids=updated_ids,
        context=context,
    )

    return await self.get(context=context)

remove_relationship async

remove_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Remove a card -> page relationship by removing the card from a page.

Source code in src/crew_dcs/classes/DomoCard/manager_page.py
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
async def remove_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Remove a card -> page relationship by removing the card from a page."""
    if relationship_type != RelationshipType.CONTAINS:
        raise ValueError(
            "Only CONTAINS relationships are supported by DomoCard_PageManager"
        )

    if target_entity_type != EntityType.PAGE:
        raise ValueError(
            "Only PAGE relationships are supported by DomoCard_PageManager"
        )

    await self.remove_pages([target_entity_id], **kwargs)
    return True
DomoLineageLink_Card(
    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(
    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/DomoCard/lineage.py
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
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 DomoCard

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

    return await DomoCard.get_by_id(
        card_id=self.id,
        auth=self.auth,
        context=context,
    )

DomoLineage_Card dataclass

DomoLineage_Card(
    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(),
)

Bases: DomoLineage

Lineage handler for card entities.

Overrides _resolve_publisher_entity to handle indirect publication: cards that are published as part of a page rather than directly.

validate_datasources

validate_datasources() -> dict[str, Any]

Validate that this card has at most one datasource.

CustomApp/EnterpriseApp cards (identified by having a datastore_id) are allowed to have multiple datasources. Regular cards should have at most one.

Returns:

Type Description
dict[str, Any]

Dictionary with validation results:

dict[str, Any]
  • valid: bool - True if card has valid datasource count
dict[str, Any]
  • datasource_count: int - Number of datasources
dict[str, Any]
  • is_custom_app: bool - True if card is part of CustomApp/EnterpriseApp
dict[str, Any]
  • violation: dict | None - Violation details if invalid, None otherwise
Source code in src/crew_dcs/classes/DomoCard/lineage.py
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def validate_datasources(self) -> dict[str, Any]:
    """Validate that this card has at most one datasource.

    CustomApp/EnterpriseApp cards (identified by having a datastore_id) are allowed
    to have multiple datasources. Regular cards should have at most one.

    Returns:
        Dictionary with validation results:
        - valid: bool - True if card has valid datasource count
        - datasource_count: int - Number of datasources
        - is_custom_app: bool - True if card is part of CustomApp/EnterpriseApp
        - violation: dict | None - Violation details if invalid, None otherwise
    """
    # Count DATA_SOURCE items in lineage
    datasource_count = sum(1 for item in self.lineage if item.type == "DATA_SOURCE")

    # Check if card is part of CustomApp/EnterpriseApp
    # Cards with datastore_id are CustomApp/EnterpriseApp cards
    is_custom_app = bool(getattr(self.parent, "datastore_id", None))

    # CustomApp/EnterpriseApp cards can have multiple datasources
    # Regular cards should have at most one
    max_allowed = float("inf") if is_custom_app else 1
    is_valid = datasource_count <= max_allowed

    result = {
        "valid": is_valid,
        "datasource_count": datasource_count,
        "is_custom_app": is_custom_app,
        "violation": None,
    }

    if not is_valid:
        # Use entity_name property if available
        if hasattr(self.parent, "entity_name"):
            card_name = self.parent.entity_name
        else:
            card_name = (
                getattr(self.parent, "title", None)
                or getattr(self.parent, "name", None)
                or str(self.parent.id)
            )

        result["violation"] = {
            "card_id": self.parent.id,
            "card_name": card_name,
            "datasource_count": datasource_count,
            "max_allowed": max_allowed,
            "datasources": [
                {
                    "id": item.id,
                    "name": (
                        item.entity.entity_name
                        if item.entity and hasattr(item.entity, "entity_name")
                        else (
                            getattr(item.entity, "name", item.id)
                            if item.entity
                            else item.id
                        )
                    ),
                }
                for item in self.lineage
                if item.type == "DATA_SOURCE"
            ],
        }

    return result

DonutCardBuilder dataclass

DonutCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Donut chart (dimension + measure).

Useful for revenue mix, cost structure visualization.

Filter dataclass

Filter(
    column: str | None = None,
    values: list[Any] = list(),
    filter_type: str | None = None,
    operand: str | None = None,
    data_type: str | None = None,
    extras: dict[str, Any] = dict(),
)

A filter applied to a subscription.

The column field can be a raw column name or a formula ID (starting with "calculation_").

Attributes:

Name Type Description
column str | None

Column name or formula ID

values list[Any]

Filter values

filter_type str | None

Filter type (e.g., "LEGACY")

operand str | None

Filter operand (e.g., "IN", "NOT_IN", "EQUALS")

data_type str | None

Data type of the column (e.g., "string", "numeric")

is_formula property

is_formula: bool

Whether this filter references a beast mode.

to_dict

to_dict() -> dict[str, Any]

Reconstruct the API dict this Filter was parsed from.

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def to_dict(self) -> dict[str, Any]:
    """Reconstruct the API dict this Filter was parsed from."""
    out: dict[str, Any] = {}
    if self.column is not None:
        out["column"] = self.column
    out["values"] = self.values
    if self.filter_type is not None:
        out["filterType"] = self.filter_type
    if self.operand is not None:
        out["operand"] = self.operand
    if self.data_type is not None:
        out["dataType"] = self.data_type
    out.update(self.extras)
    return out

Formula dataclass

Formula(
    id: str = "",
    name: str = "",
    formula: str = "",
    resolved_formula: str | None = None,
    status: str | None = None,
    data_type: str | None = None,
    column_positions: list[ColumnRef] = list(),
    non_aggregated_columns: list[str] = list(),
    template_id: int | None = None,
    variable: bool = False,
    persisted_on_data_source: bool = False,
    used_by_other_cards: bool = False,
    reference_count: int = 0,
    is_controlled: bool = False,
    is_aggregatable: bool = True,
)

A beast mode formula in a card's definition.

Attributes:

Name Type Description
id str

Unique formula ID (e.g., "calculation_272e5ef4-...")

name str

Display name

formula str

SQL expression

status str | None

Validation status (e.g., "VALID")

data_type str | None

Output data type (e.g., "DECIMAL", "STRING", "LONG")

column_positions list[ColumnRef]

Columns referenced in the formula

non_aggregated_columns list[str]

Columns used without aggregation

template_id int | None

Formula template ID

variable bool

Whether this is a variable

persisted_on_data_source bool

Whether persisted to the dataset

used_by_other_cards bool

Whether other cards reference this formula

reference_count int

Number of cards referencing this formula

is_controlled bool

Whether this is a controlled formula

is_aggregatable bool

Whether the formula result is aggregatable

referenced_columns property

referenced_columns: set[str]

All dataset columns referenced by this formula.

Combines columnPositions and nonAggregatedColumns, stripping backticks from column names.

to_dict

to_dict() -> dict[str, Any]

Reconstruct the API dict for this formula (typed subset).

NOTE: the parser reads only a subset of a beast mode's fields, so this is intentionally lossy relative to the original payload (fields such as cacheWindow, locked, owner, isAnalytic, bignumber are not modeled and are therefore not emitted).

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
def to_dict(self) -> dict[str, Any]:
    """Reconstruct the API dict for this formula (typed subset).

    NOTE: the parser reads only a subset of a beast mode's fields, so this
    is intentionally lossy relative to the original payload (fields such as
    ``cacheWindow``, ``locked``, ``owner``, ``isAnalytic``, ``bignumber``
    are not modeled and are therefore not emitted).
    """
    out: dict[str, Any] = {
        "id": self.id,
        "name": self.name,
        "formula": self.formula,
        "variable": self.variable,
        "persistedOnDataSource": self.persisted_on_data_source,
        "usedByOtherCards": self.used_by_other_cards,
        "referenceCount": self.reference_count,
        "isControlled": self.is_controlled,
        "isAggregatable": self.is_aggregatable,
    }
    if self.status is not None:
        out["status"] = self.status
    if self.data_type is not None:
        out["dataType"] = self.data_type
    if self.template_id is not None:
        out["templateId"] = self.template_id
    if self.column_positions:
        out["columnPositions"] = [cp.to_dict() for cp in self.column_positions]
    if self.non_aggregated_columns:
        out["nonAggregatedColumns"] = self.non_aggregated_columns
    return out

FunnelCardBuilder dataclass

FunnelCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Funnel chart (dimension + measure). EXPERIMENTAL: badge_funnel is unverified against a live instance.

Useful as a P&L waterfall alternative: Revenue -> COGS -> Gross Profit -> Operating Expenses -> Net Income. Each stage is a dimension value.

GaugeCardBuilder dataclass

GaugeCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Filled-gauge chart (single measure, no dimension).

Useful for liquidity ratios (Current Ratio, Debt-to-Equity) and other single-value KPIs that benefit from a visual range indicator. Uses badge_filledgauge (a chartType confirmed present on live instances); badge_compgauge is the comparative-gauge alternative.

GroupByColumn dataclass

GroupByColumn(
    column: str | None = None,
    formula_id: str | None = None,
    extras: dict[str, Any] = dict(),
)

A group-by column in a subscription.

Attributes:

Name Type Description
column str | None

Raw column name (or None if formula_id)

formula_id str | None

Beast mode calculation ID (or None if raw column)

to_dict

to_dict() -> dict[str, Any]

Reconstruct the API dict this GroupByColumn was parsed from.

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
312
313
314
315
316
317
318
319
320
def to_dict(self) -> dict[str, Any]:
    """Reconstruct the API dict this GroupByColumn was parsed from."""
    out: dict[str, Any] = {}
    if self.column is not None:
        out["column"] = self.column
    if self.formula_id is not None:
        out["formulaId"] = self.formula_id
    out.update(self.extras)
    return out

GroupedBarCardBuilder dataclass

GroupedBarCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Vertical grouped/multi bar chart (dimension + measure + series).

HeatmapCardBuilder dataclass

HeatmapCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Heatmap chart (dimension + measure + series).

Useful for period-over-period comparison (e.g. monthly revenue by account type across years).

HorizontalBarCardBuilder dataclass

HorizontalBarCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Horizontal bar chart (dimension + measure).

Useful for P&L line items where account names are long and read better on the y-axis.

HorizontalGroupedBarCardBuilder dataclass

HorizontalGroupedBarCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Horizontal grouped/multi bar chart (dimension + measure + series). EXPERIMENTAL: badge_horiz_multibar is unverified; the confirmed grouped-bar type is the vertical badge_vert_multibar (grouped_bar).

Useful for Actual vs Budget comparisons by account category.

HorizontalStackedBarCardBuilder dataclass

HorizontalStackedBarCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Horizontal stacked bar chart (dimension + measure + series).

Useful for balance sheet composition (Assets/Liabilities/Equity stacked by account type).

KpiCardBuilder dataclass

KpiCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Base builder for a Domo KPI card definition.

Subclasses register themselves with :func:register_card_type, which sets the friendly_name and chart_type class attributes. Behavioural flags (requires_dimension, supports_series, is_single_value) control how the main subscription is assembled.

Attributes:

Name Type Description
dataset_id str

Source dataset id. Retained for the caller to bind at POST time; NOT embedded in the emitted definition.

title str

Card title.

measure_column str | None

The VALUE column (aggregated numeric measure).

aggregation str

Aggregation applied to measure_column (e.g. "SUM").

dimension_column str | None

The ITEM / primary groupBy column.

series_column str | None

Optional SERIES / secondary grouping column.

filters list[Any]

Optional filters (raw dicts or :class:Filter instances).

order_by list[Any]

Optional sorts (raw dicts or :class:SortColumn instances).

overrides dict[str, Any]

Chart overrides (e.g. title_x, title_y, footer).

description str | None

Optional card description.

allow_table_drill bool

Whether table drill is allowed (default True).

build

build() -> KpiDefinition

Assemble and return a fully-typed :class:KpiDefinition.

Source code in src/crew_dcs/classes/DomoCard/kpi_builder.py
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
def build(self) -> KpiDefinition:
    """Assemble and return a fully-typed :class:`KpiDefinition`."""
    self._validate()

    # Build the beast-mode measure (if any) before assembling columns so the
    # VALUE mapping can reference it by formula id.
    formulas: list[Formula] = []
    self._formula_id = None
    if self.measure_beastmode:
        ref = self.measure_beastmode
        fobj = build_persisted_formula(
            legacy_id=ref["legacy_id"],
            template_id=ref["template_id"],
            name=ref.get("name", ""),
            expression=ref["expression"],
            data_type=ref.get("data_type", "DECIMAL"),
        )
        self._formula_id = fobj.id
        formulas.append(fobj)
    elif self.measure_formula:
        fobj = build_formula(
            name=self.measure_formula["name"],
            expression=self.measure_formula["formula"],
            data_type=self.measure_formula.get("data_type", "DECIMAL"),
        )
        self._formula_id = fobj.id
        formulas.append(fobj)

    main_extras: dict[str, Any] = {}
    if self.date_grain:
        main_extras["dateGrain"] = self.date_grain

    main = Subscription(
        name="main",
        columns=self.build_columns(),
        filters=self._coerce_filters(),
        order_by=self._coerce_order_by(),
        group_by=self.build_group_by(),
        fiscal=False,
        projection=False,
        distinct=False,
        extras=main_extras,
    )

    subscriptions: dict[str, Subscription] = {"main": main}

    # Optional big_number subscription (combo cards: summary number above chart)
    if self.big_number_column or self.big_number_formula_id:
        bn_col: ColumnMapping
        if self.big_number_formula_id:
            bn_col = ColumnMapping(
                formula_id=self.big_number_formula_id,
                mapping="VALUE",
            )
        else:
            bn_col = ColumnMapping(
                column=self.big_number_column,
                aggregation=self.big_number_aggregation,
                mapping="VALUE",
            )
        big_number = Subscription(
            name="big_number",
            columns=[bn_col],
            filters=self._coerce_filters(),
            order_by=[],
            group_by=[],
            fiscal=False,
            projection=False,
            distinct=False,
        )
        subscriptions["big_number"] = big_number

    charts = {
        "main": {
            "component": "main",
            "chartType": self.chart_type,
            "overrides": dict(self.overrides),
            "goal": None,
        }
    }

    return KpiDefinition(
        subscriptions=subscriptions,
        formulas=formulas,
        charts=charts,
        segments={"active": [], "definitions": []},
        conditional_formats=[],
        annotations=[],
        slicers=[],
        title=self.title,
        description=self.description,
        chart_version="12",
        allow_table_drill=self.allow_table_drill,
        input_table=False,
    )

build_columns

build_columns() -> list[ColumnMapping]

Build the main subscription column mappings.

Source code in src/crew_dcs/classes/DomoCard/kpi_builder.py
316
317
318
319
320
321
322
323
324
325
326
327
def build_columns(self) -> list[ColumnMapping]:
    """Build the ``main`` subscription column mappings."""
    if self.is_single_value:
        return [self._measure_mapping("VALUE")]

    columns = [
        ColumnMapping(column=self.dimension_column, mapping="ITEM"),
        self._measure_mapping("VALUE"),
    ]
    if self.supports_series and self.series_column:
        columns.append(ColumnMapping(column=self.series_column, mapping="SERIES"))
    return columns

build_group_by

build_group_by() -> list[GroupByColumn]

Build the main subscription groupBy columns.

Source code in src/crew_dcs/classes/DomoCard/kpi_builder.py
329
330
331
332
333
334
335
336
def build_group_by(self) -> list[GroupByColumn]:
    """Build the ``main`` subscription groupBy columns."""
    if self.is_single_value or not self.dimension_column:
        return []
    groups = [GroupByColumn(column=self.dimension_column)]
    if self.supports_series and self.series_column:
        groups.append(GroupByColumn(column=self.series_column))
    return groups

to_dict

to_dict() -> dict[str, Any]

Build the card and return the full create/definition envelope.

Source code in src/crew_dcs/classes/DomoCard/kpi_builder.py
459
460
461
def to_dict(self) -> dict[str, Any]:
    """Build the card and return the full create/definition envelope."""
    return self.build().to_dict()

KpiDefinition dataclass

KpiDefinition(
    subscriptions: dict[str, Subscription] = dict(),
    formulas: list[Formula] = list(),
    charts: dict[str, Any] = dict(),
    segments: dict[str, Any] = dict(),
    conditional_formats: list[dict] = list(),
    annotations: list[dict] = list(),
    slicers: list[dict] = list(),
    title: str | None = None,
    description: str | None = None,
    chart_version: str | None = None,
    allow_table_drill: bool = False,
    input_table: bool = False,
    modified: int | None = None,
    columns_schema: list[ColumnSchema] = list(),
)

Typed representation of a card's kpi definition.

Provides structured access to subscriptions, formulas, and column schema with convenience properties for common queries.

Usage

kpi = KpiDefinition.from_dict(api_response) kpi.main.columns # list[ColumnMapping] kpi.formula_by_id[...] # Formula lookup kpi.used_columns # set[str] of all columns used kpi.available_columns # list[ColumnSchema] of all dataset columns

available_columns property

available_columns: list[str]

All available column names from the dataset schema.

beast_modes property

beast_modes: list[Formula]

All beast mode formulas (non-variable calculations).

definition property

definition: dict[str, Any]

The inner definition dict of the create/GET envelope.

formula_by_id property

formula_by_id: dict[str, Formula]

Lookup dict of formulas by their calculation ID.

main property

The main subscription (most common use case).

unused_columns property

unused_columns: set[str]

Dataset columns NOT used by this card.

used_columns property

used_columns: set[str]

All dataset columns used by this card.

Resolves formula references to their underlying columns. Combines columns from: - Subscription column mappings - Subscription filters - Subscription orderBy - Subscription groupBy - All formula columnPositions and nonAggregatedColumns

from_dict classmethod

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

Build a KpiDefinition from the kpi definition API response.

Parameters:

Name Type Description Default
obj dict[str, Any]

Full API response dict (includes top-level 'definition' and 'columns')

required
Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
@classmethod
def from_dict(cls, obj: dict[str, Any]) -> KpiDefinition:
    """Build a KpiDefinition from the kpi definition API response.

    Args:
        obj: Full API response dict (includes top-level 'definition' and 'columns')
    """
    definition = obj.get("definition", obj)

    # Parse subscriptions
    subscriptions = {
        name: Subscription.from_dict(sub_data)
        for name, sub_data in definition.get("subscriptions", {}).items()
        if isinstance(sub_data, dict)
    }

    # Parse formulas
    formulas = [Formula.from_dict(f) for f in definition.get("formulas", [])]

    # Parse column schema (top-level 'columns' in API response)
    columns_schema = [ColumnSchema.from_dict(c) for c in obj.get("columns", [])]

    return cls(
        subscriptions=subscriptions,
        formulas=formulas,
        charts=definition.get("charts", {}),
        segments=definition.get("segments", {}),
        conditional_formats=definition.get("conditionalFormats", []),
        annotations=definition.get("annotations", []),
        slicers=definition.get("slicers", []),
        title=definition.get("title"),
        description=definition.get("description"),
        chart_version=definition.get("chartVersion"),
        allow_table_drill=definition.get("allowTableDrill", False),
        input_table=definition.get("inputTable", False),
        modified=definition.get("modified"),
        columns_schema=columns_schema,
    )

resolve_domo_beast_mode_refs async

resolve_domo_beast_mode_refs(
    auth: Any, *, context: Any = None, **context_kwargs
) -> None

Resolve DOMO_BEAST_MODE(nnn) references in all formulas.

Domo beast modes can reference other beast modes using the syntax DOMO_BEAST_MODE(template_id). This method fetches the referenced templates and populates resolved_formula on each Formula that contains such references.

Parameters:

Name Type Description Default
auth Any

Authentication object for API requests

required
context Any

Optional RouteContext for request configuration

None
Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
async def resolve_domo_beast_mode_refs(
    self,
    auth: Any,
    *,
    context: Any = None,
    **context_kwargs,
) -> None:
    """Resolve DOMO_BEAST_MODE(nnn) references in all formulas.

    Domo beast modes can reference other beast modes using the syntax
    ``DOMO_BEAST_MODE(template_id)``. This method fetches the referenced
    templates and populates ``resolved_formula`` on each Formula that
    contains such references.

    Args:
        auth: Authentication object for API requests
        context: Optional RouteContext for request configuration
    """
    import re
    from ...routes import beastmode as beastmode_routes
    from ...client.context import RouteContext

    _DOMO_BM_RE = re.compile(r"DOMO_BEAST_MODE\((\d+)\)")
    context = RouteContext.build_context(context=context, **context_kwargs)

    # Collect all unique template IDs referenced across all formulas
    all_refs: dict[int, Any] = {}  # template_id -> BeastModeTemplate
    for formula in self.formulas:
        for match in _DOMO_BM_RE.findall(formula.formula or ""):
            tid = int(match)
            if tid not in all_refs:
                all_refs[tid] = None

    if not all_refs:
        return

    # Fetch all referenced templates
    for tid in all_refs:
        try:
            res = await beastmode_routes.get_beastmode_by_id(
                auth=auth,
                beastmode_id=str(tid),
                include_hidden=True,
                context=context,
            )
            if res.is_success:
                from ..DomoBeastMode.template_definition import BeastModeTemplate

                all_refs[tid] = BeastModeTemplate.from_dict(res.response)
        except Exception:  # noqa: BLE001
            pass

    # Populate resolved_formula on each formula
    for formula in self.formulas:
        refs_in_formula = _DOMO_BM_RE.findall(formula.formula or "")
        if not refs_in_formula:
            continue
        resolved = formula.formula
        for tid_str in refs_in_formula:
            tid = int(tid_str)
            template = all_refs.get(tid)
            if template:
                resolved = resolved.replace(
                    f"DOMO_BEAST_MODE({tid})",
                    f"({template.expression})",
                )
        formula.resolved_formula = resolved

resolve_unresolved_formulas

resolve_unresolved_formulas(dataset_formulas: Any) -> None

Resolve formula IDs that aren't in the card's own formulas list.

Card definitions only include card-level beast modes. Dataset-level beast modes are referenced by their calculation_xxx legacy ID but their definitions live in the template API. This method resolves them using the dataset's Formulas manager.

After calling this, formula_by_id will include dataset-level formulas, and used_columns will correctly resolve them.

Parameters:

Name Type Description Default
dataset_formulas Any

DomoDataset_FormulasManager with loaded beast modes

required
Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
def resolve_unresolved_formulas(self, dataset_formulas: Any) -> None:  # noqa: C901
    """Resolve formula IDs that aren't in the card's own formulas list.

    Card definitions only include card-level beast modes. Dataset-level
    beast modes are referenced by their ``calculation_xxx`` legacy ID
    but their definitions live in the template API. This method resolves
    them using the dataset's Formulas manager.

    After calling this, ``formula_by_id`` will include dataset-level
    formulas, and ``used_columns`` will correctly resolve them.

    Args:
        dataset_formulas: DomoDataset_FormulasManager with loaded beast modes
    """
    if not dataset_formulas or not dataset_formulas.beast_modes:
        return

    # Build lookup from dataset-level beast modes
    dataset_lookup = dataset_formulas.formula_by_legacy_id
    if not dataset_lookup:
        return

    existing_ids = {f.id for f in self.formulas}

    # Find unresolved formula IDs in subscriptions
    unresolved_ids: set[str] = set()

    for sub in self.subscriptions.values():
        for mapping in sub.columns:
            if mapping.formula_id and mapping.formula_id not in existing_ids:
                unresolved_ids.add(mapping.formula_id)

        for gb in sub.group_by:
            if gb.formula_id and gb.formula_id not in existing_ids:
                unresolved_ids.add(gb.formula_id)

        for o in sub.order_by:
            if o.formula_id and o.formula_id not in existing_ids:
                unresolved_ids.add(o.formula_id)

        for f in sub.filters:
            if (
                f.column
                and f.column.startswith("calculation_")
                and f.column not in existing_ids
            ):
                unresolved_ids.add(f.column)

    # Resolve and add
    for calc_id in unresolved_ids:
        template = dataset_lookup.get(calc_id)
        if template:
            self.formulas.append(template.to_formula())

to_dict

to_dict() -> dict[str, Any]

Reconstruct the full create/definition envelope.

Returns the {"definition": {...}, "columns": [...]} envelope.

Important: The Domo Content API has different formats for GET vs CREATE (PUT /content/v3/cards/kpi):

  • GET (read existing card): formulas, annotations, and conditionalFormats are returned as arrays.
  • CREATE (PUT new card): formulas, annotations, and conditionalFormats must be objects with change-tracking sub-keys:

  • formulas: {"dsUpdated": [], "dsDeleted": [], "card": [...]}

  • annotations: {"new": [], "modified": [], "deleted": []}
  • conditionalFormats: {"card": [...], "datasource": []}

This method produces the CREATE format. If you need the GET format (e.g. for round-trip comparison), use :meth:to_get_dict.

The subscriptions, charts and "chrome" fields round-trip exactly; the formulas and dataset columns are reconstructed from the typed subset the parser reads and are therefore lossy (see Formula.to_dict and ColumnSchema.to_dict).

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
def to_dict(self) -> dict[str, Any]:
    """Reconstruct the full create/definition envelope.

    Returns the ``{"definition": {...}, "columns": [...]}`` envelope.

    **Important**: The Domo Content API has different formats for GET vs
    CREATE (PUT /content/v3/cards/kpi):

    - **GET** (read existing card): ``formulas``, ``annotations``, and
      ``conditionalFormats`` are returned as **arrays**.
    - **CREATE** (PUT new card): ``formulas``, ``annotations``, and
      ``conditionalFormats`` must be **objects** with change-tracking
      sub-keys:

      - ``formulas``: ``{"dsUpdated": [], "dsDeleted": [], "card": [...]}``
      - ``annotations``: ``{"new": [], "modified": [], "deleted": []}``
      - ``conditionalFormats``: ``{"card": [...], "datasource": []}``

    This method produces the **CREATE** format.  If you need the GET format
    (e.g. for round-trip comparison), use :meth:`to_get_dict`.

    The subscriptions, charts and "chrome" fields round-trip exactly; the
    ``formulas`` and dataset ``columns`` are reconstructed from the typed
    subset the parser reads and are therefore lossy (see ``Formula.to_dict``
    and ``ColumnSchema.to_dict``).
    """
    definition: dict[str, Any] = {
        "subscriptions": {
            name: sub.to_dict() for name, sub in self.subscriptions.items()
        },
        # CREATE format: formulas/annotations/conditionalFormats are objects
        # with change-tracking sub-keys, NOT arrays.
        "formulas": {
            "dsUpdated": [],
            "dsDeleted": [],
            "card": [f.to_dict() for f in self.formulas],
        },
        "conditionalFormats": {
            "card": self.conditional_formats,
            "datasource": [],
        },
        "annotations": {
            "new": self.annotations,
            "modified": [],
            "deleted": [],
        },
        "slicers": self.slicers,
        "title": self.title,
        "chartVersion": self.chart_version,
        "charts": self.charts,
        "allowTableDrill": self.allow_table_drill,
        "segments": self.segments,
        "inputTable": self.input_table,
    }
    if self.description is not None:
        definition["description"] = self.description
    # Do NOT include 'modified' on create — it's server-assigned.

    return {
        "definition": definition,
        "columns": [c.to_dict() for c in self.columns_schema],
    }

to_get_dict

to_get_dict() -> dict[str, Any]

Reconstruct the GET-format envelope (arrays for formulas/annotations/conditionalFormats).

Use this when you want to compare against the raw API GET response. For card creation, use :meth:to_dict (the CREATE format).

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
def to_get_dict(self) -> dict[str, Any]:
    """Reconstruct the GET-format envelope (arrays for formulas/annotations/conditionalFormats).

    Use this when you want to compare against the raw API GET response.
    For card creation, use :meth:`to_dict` (the CREATE format).
    """
    definition: dict[str, Any] = {
        "subscriptions": {
            name: sub.to_dict() for name, sub in self.subscriptions.items()
        },
        "formulas": [f.to_dict() for f in self.formulas],
        "conditionalFormats": self.conditional_formats,
        "annotations": self.annotations,
        "slicers": self.slicers,
        "title": self.title,
        "chartVersion": self.chart_version,
        "charts": self.charts,
        "allowTableDrill": self.allow_table_drill,
        "segments": self.segments,
        "inputTable": self.input_table,
    }
    if self.description is not None:
        definition["description"] = self.description
    if self.modified is not None:
        definition["modified"] = self.modified

    return {
        "definition": definition,
        "columns": [c.to_dict() for c in self.columns_schema],
    }

LineCardBuilder dataclass

LineCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Line/line-bar combo chart (dimension + measure, optional series).

Useful for trend over time (Revenue, Expenses, Net Income).

ParetoCardBuilder dataclass

ParetoCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Pareto chart (dimension + measure). EXPERIMENTAL: badge_pareto is unverified against a live instance.

Shows the 80/20 rule — useful for identifying which accounts or cost categories drive the majority of revenue or expense.

PieCardBuilder dataclass

PieCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Pie chart (dimension + measure).

Useful for asset allocation, expense distribution, revenue mix.

SingleValueCardBuilder dataclass

SingleValueCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Single big-number card (no dimension).

SortColumn dataclass

SortColumn(
    column: str | None = None,
    formula_id: str | None = None,
    order: str | None = None,
    aggregation: str | None = None,
    extras: dict[str, Any] = dict(),
)

A sort specification in a subscription.

Attributes:

Name Type Description
column str | None

Raw column name (or None if formula_id)

formula_id str | None

Beast mode calculation ID (or None if raw column)

order str | None

Sort direction ("ASCENDING" or "DESCENDING")

to_dict

to_dict() -> dict[str, Any]

Reconstruct the API dict this SortColumn was parsed from.

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
270
271
272
273
274
275
276
277
278
279
280
281
282
def to_dict(self) -> dict[str, Any]:
    """Reconstruct the API dict this SortColumn was parsed from."""
    out: dict[str, Any] = {}
    if self.column is not None:
        out["column"] = self.column
    if self.formula_id is not None:
        out["formulaId"] = self.formula_id
    if self.aggregation is not None:
        out["aggregation"] = self.aggregation
    if self.order is not None:
        out["order"] = self.order
    out.update(self.extras)
    return out

StackedBarCardBuilder dataclass

StackedBarCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Vertical stacked bar chart (dimension + measure + series).

Subscription dataclass

Subscription(
    name: str = "",
    columns: list[ColumnMapping] = list(),
    filters: list[Filter] = list(),
    order_by: list[SortColumn] = list(),
    group_by: list[GroupByColumn] = list(),
    fiscal: bool = False,
    projection: bool = False,
    distinct: bool = False,
    extras: dict[str, Any] = dict(),
)

A subscription within a kpi definition (typically "main").

Attributes:

Name Type Description
name str

Subscription name (e.g., "main")

columns list[ColumnMapping]

Column-to-axis mappings

filters list[Filter]

Applied filters

order_by list[SortColumn]

Sort specifications

group_by list[GroupByColumn]

Group-by columns

fiscal bool

Whether fiscal calendar is used

projection bool

Whether projection is enabled

distinct bool

Whether DISTINCT is applied

to_dict

to_dict() -> dict[str, Any]

Reconstruct the API dict this Subscription was parsed from.

Preserves unmodeled subscription-level keys (limit, dateGrain, dateRangeFilter, ...) captured in extras so the round-trip is exact.

Source code in src/crew_dcs/classes/DomoCard/kpi_definition.py
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
def to_dict(self) -> dict[str, Any]:
    """Reconstruct the API dict this Subscription was parsed from.

    Preserves unmodeled subscription-level keys (``limit``, ``dateGrain``,
    ``dateRangeFilter``, ...) captured in ``extras`` so the round-trip is
    exact.
    """
    out: dict[str, Any] = {
        "name": self.name,
        "columns": [c.to_dict() for c in self.columns],
        "filters": [f.to_dict() for f in self.filters],
        "orderBy": [o.to_dict() for o in self.order_by],
        "groupBy": [g.to_dict() for g in self.group_by],
        "fiscal": self.fiscal,
        "projection": self.projection,
        "distinct": self.distinct,
    }
    out.update(self.extras)
    return out

TableCardBuilder dataclass

TableCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Basic table card.

A table lists one or more VALUE columns and does not require a dimension / groupBy. The base build_columns emits an ITEM + VALUE pair; for a table we emit the measure as a plain VALUE column with no groupBy.

TreemapCardBuilder dataclass

TreemapCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Treemap chart (dimension + measure + series). EXPERIMENTAL: badge_treemap is unverified; badge_sunburst is the confirmed hierarchical alternative.

Useful for hierarchical financial data — shows account hierarchy as nested rectangles sized by amount.

TrendlineCardBuilder dataclass

TrendlineCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Trend line chart (dimension + measure). EXPERIMENTAL: badge_trendline is unverified; confirmed line-ish alternatives are badge_rttrendline, badge_two_trendline and badge_symbolline.

Useful for showing financial trends over time (revenue growth, cost trends, margin trajectory).

WaterfallCardBuilder dataclass

WaterfallCardBuilder(
    dataset_id: str = "",
    title: str = "",
    measure_column: str | None = None,
    aggregation: str = "SUM",
    measure_formula: dict[str, Any] | None = None,
    measure_beastmode: dict[str, Any] | None = None,
    dimension_column: str | None = None,
    series_column: str | None = None,
    filters: list[Any] = list(),
    order_by: list[Any] = list(),
    overrides: dict[str, Any] = dict(),
    description: str | None = None,
    allow_table_drill: bool = True,
    date_grain: dict[str, Any] | None = None,
    big_number_column: str | None = None,
    big_number_aggregation: str = "SUM",
    big_number_formula_id: str | None = None,
)

Bases: KpiCardBuilder

Waterfall-style chart using vertical bar (dimension + measure).

Domo does not expose a dedicated badge_waterfall chartType via the Content API. This builder uses badge_vert_bar as the closest substitute — the card can be switched to waterfall display in the Domo UI after creation. Each dimension value becomes a step in the P&L bridge (Revenue -> COGS -> Gross Profit -> Net Income).

build_kpi_card_definition

build_kpi_card_definition(
    chart_type: str, **kwargs: Any
) -> dict[str, Any]

Build a card definition envelope for a friendly chart type.

Parameters:

Name Type Description Default
chart_type str

Friendly chart-type name (see :func:get_registered_card_types).

required
**kwargs Any

Passed through to the :class:KpiCardBuilder subclass (dataset_id, title, measure_column, aggregation, dimension_column, series_column, filters, order_by, overrides, description, ...).

{}

Returns:

Type Description
dict[str, Any]

The full envelope {"definition": {...}, "columns": [...]}. The

dict[str, Any]

ready-to-send inner definition is available at result["definition"].

Source code in src/crew_dcs/classes/DomoCard/kpi_builder.py
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
def build_kpi_card_definition(chart_type: str, **kwargs: Any) -> dict[str, Any]:
    """Build a card definition envelope for a friendly chart type.

    Args:
        chart_type: Friendly chart-type name (see :func:`get_registered_card_types`).
        **kwargs: Passed through to the :class:`KpiCardBuilder` subclass
            (``dataset_id``, ``title``, ``measure_column``, ``aggregation``,
            ``dimension_column``, ``series_column``, ``filters``, ``order_by``,
            ``overrides``, ``description``, ...).

    Returns:
        The full envelope ``{"definition": {...}, "columns": [...]}``.  The
        ready-to-send inner definition is available at ``result["definition"]``.
    """
    builder_cls = get_card_builder_class(chart_type)
    builder = builder_cls(**kwargs)
    return builder.to_dict()

card_type_map

card_type_map() -> dict[str, str]

Return the friendly-name -> Domo-chartType mapping.

Source code in src/crew_dcs/classes/DomoCard/kpi_builder.py
225
226
227
def card_type_map() -> dict[str, str]:
    """Return the friendly-name -> Domo-chartType mapping."""
    return dict(_CARD_CHART_TYPE_MAP)

get_card_builder_class

get_card_builder_class(
    friendly_name: str,
) -> type[KpiCardBuilder]

Return the builder class registered for a friendly chart-type name.

Raises:

Type Description
ValueError

If the friendly name is not registered.

Source code in src/crew_dcs/classes/DomoCard/kpi_builder.py
192
193
194
195
196
197
198
199
200
201
202
203
204
def get_card_builder_class(friendly_name: str) -> type[KpiCardBuilder]:
    """Return the builder class registered for a friendly chart-type name.

    Raises:
        ValueError: If the friendly name is not registered.
    """
    cls = _CARD_BUILDER_REGISTRY.get(friendly_name)
    if cls is None:
        known = ", ".join(sorted(_CARD_BUILDER_REGISTRY))
        raise ValueError(
            f"Unknown card chart type '{friendly_name}'. Registered: {known}"
        )
    return cls

get_registered_card_types

get_registered_card_types() -> list[str]

Return all registered friendly chart-type names.

Source code in src/crew_dcs/classes/DomoCard/kpi_builder.py
207
208
209
def get_registered_card_types() -> list[str]:
    """Return all registered friendly chart-type names."""
    return sorted(_CARD_BUILDER_REGISTRY)

register_card_type

register_card_type(friendly_name: str, chart_type: str)

Decorator registering a :class:KpiCardBuilder subclass.

Parameters:

Name Type Description Default
friendly_name str

Human-friendly key (e.g. "bar", "single_value").

required
chart_type str

Domo chartType string (e.g. "badge_vert_bar").

required
Source code in src/crew_dcs/classes/DomoCard/kpi_builder.py
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def register_card_type(friendly_name: str, chart_type: str):
    """Decorator registering a :class:`KpiCardBuilder` subclass.

    Args:
        friendly_name: Human-friendly key (e.g. ``"bar"``, ``"single_value"``).
        chart_type: Domo ``chartType`` string (e.g. ``"badge_vert_bar"``).
    """

    def decorator(cls: type[KpiCardBuilder]) -> type[KpiCardBuilder]:
        cls.friendly_name = friendly_name
        cls.chart_type = chart_type
        _CARD_BUILDER_REGISTRY[friendly_name] = cls
        _CARD_CHART_TYPE_MAP[friendly_name] = chart_type
        return cls

    return decorator

Modules