Skip to content

DomoObjective

DomoObjective

DomoObjective Package

Class-based approach for interacting with Domo Goals (Objectives), key results, and objective tags.

Modules:

Name Description
core

DomoObjective entity and DomoObjectives manager
key_result

DomoKeyResult and DomoKeyResults sub-entity
tag

DomoObjectiveTag and DomoObjectiveTags sub-entity

DomoKeyResult dataclass 

DomoKeyResult(
    id: int,
    name: str = None,
    description: str = None,
    state: str = None,
    status: str = None,
    owner_id: int = None,
    owner_type: str = None,
    owners: list[Any] = None,
    resource_id: str = None,
    resource_type: str = None,
    measurement: str = None,
    start_value: float = None,
    current_value: float = None,
    target_value: float = None,
    color_value: str = None,
    operator: str = None,
    raw: dict = None,
)

Represents a single key result belonging to an objective.

DomoKeyResults dataclass 

DomoKeyResults(
    parent: DomoEntity,
    key_results: list[DomoKeyResult] = list(),
)

Bases: DomoSubEntity

Sub-entity for managing an objective's key results.

create async 

create(
    name: str,
    owner_id: int,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoKeyResult

Create a new key result.

Source code in src/crew_dcs/classes/DomoObjective/key_result.py 
 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
async def create(
    self,
    name: str,
    owner_id: int,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoKeyResult:
    """Create a new key result."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.create_key_result(
        auth=self.auth,
        name=name,
        owner_id=owner_id,
        context=context,
    )

    if return_raw:
        return res

    key_result = DomoKeyResult.from_dict(res.response)
    self.key_results.append(key_result)
    return key_result

delete async 

delete(
    key_result_id: int | str,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Delete a key result.

Source code in src/crew_dcs/classes/DomoObjective/key_result.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
async def delete(
    self,
    key_result_id: int | str,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Delete a key result."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.delete_key_result(
        auth=self.auth,
        key_result_id=key_result_id,
        context=context,
    )

    self.key_results = [
        kr for kr in self.key_results if str(kr.id) != str(key_result_id)
    ]
    return res

get async 

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

Populate key results from the parent objective's raw data.

Source code in src/crew_dcs/classes/DomoObjective/key_result.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def get(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoKeyResult]:
    """Populate key results from the parent objective's raw data."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    raw = getattr(self.parent, "raw", None) or {}
    self.key_results = [
        DomoKeyResult.from_dict(obj) for obj in raw.get("keyResults", [])
    ]
    return self.key_results

get_chart async 

get_chart(
    key_result_id: int | str,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Get chart data for a key result.

Source code in src/crew_dcs/classes/DomoObjective/key_result.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
async def get_chart(
    self,
    key_result_id: int | str,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Get chart data for a key result."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await objectives_routes.get_key_result_chart(
        auth=self.auth,
        key_result_id=key_result_id,
        context=context,
    )

get_values async 

get_values(
    key_result_id: int | str,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Get historical values for a key result.

Source code in src/crew_dcs/classes/DomoObjective/key_result.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
async def get_values(
    self,
    key_result_id: int | str,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Get historical values for a key result."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await objectives_routes.get_key_result_values(
        auth=self.auth,
        key_result_id=key_result_id,
        context=context,
    )

update async 

update(
    key_result_id: int | str,
    payload: dict,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoKeyResult

Update an existing key result.

Source code in src/crew_dcs/classes/DomoObjective/key_result.py 
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
async def update(
    self,
    key_result_id: int | str,
    payload: dict,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoKeyResult:
    """Update an existing key result."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.update_key_result(
        auth=self.auth,
        key_result_id=key_result_id,
        payload=payload,
        context=context,
    )

    return DomoKeyResult.from_dict(res.response)

update_tags async 

update_tags(
    key_result_id: int | str,
    tag_ids: list[int],
    period_id: int | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Replace the tags assigned to a key result.

Source code in src/crew_dcs/classes/DomoObjective/key_result.py 
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
async def update_tags(
    self,
    key_result_id: int | str,
    tag_ids: list[int],
    period_id: int | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Replace the tags assigned to a key result."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await objectives_routes.update_key_result_tags(
        auth=self.auth,
        key_result_id=key_result_id,
        tag_ids=tag_ids,
        period_id=period_id,
        context=context,
    )

DomoObjective dataclass 

DomoObjective(
    auth: DomoAuth,
    id: int,
    raw: dict,
    name: str = None,
    description: str = None,
    starts_at: Any = None,
    expires_at: Any = None,
    status: str = None,
    owner_id: int = None,
    owners: list[Any] = None,
    assignees: list[Any] = None,
    period_id: int = None,
    parent_id: int = None,
    tags: list[Any] = None,
    write_access: bool = None,
    type: str = None,
)

Bases: DomoEntity

Represents a Domo Goal/Objective.

create async classmethod 

create(
    auth: DomoAuth,
    name: str,
    period_id: int,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoObjective

Create a new objective.

Source code in src/crew_dcs/classes/DomoObjective/core.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
@classmethod
async def create(
    cls,
    auth: DomoAuth,
    name: str,
    period_id: int,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoObjective:
    """Create a new objective."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.create_objective(
        auth=auth,
        name=name,
        period_id=period_id,
        context=context,
    )

    if return_raw:
        return res

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

delete async 

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

Delete this objective.

Source code in src/crew_dcs/classes/DomoObjective/core.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
async def delete(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Delete this objective."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await objectives_routes.delete_objective(
        auth=self.auth,
        objective_id=self.id,
        context=context,
    )

from_dict classmethod 

from_dict(
    auth: DomoAuth, obj: dict[str, Any], **kwargs
) -> DomoObjective

Converts an objectives API response into a DomoObjective instance.

Source code in src/crew_dcs/classes/DomoObjective/core.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@classmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any], **kwargs) -> DomoObjective:
    """Converts an objectives API response into a DomoObjective instance."""

    return cls(
        id=obj.get("id"),
        auth=auth,
        name=obj.get("name"),
        description=obj.get("description"),
        starts_at=cd.convert_string_to_datetime(obj.get("startsAt")),
        expires_at=cd.convert_string_to_datetime(obj.get("expiresAt")),
        status=obj.get("status"),
        owner_id=obj.get("ownerId"),
        owners=obj.get("owners"),
        assignees=obj.get("assignees"),
        period_id=obj.get("periodId"),
        parent_id=obj.get("parentId"),
        tags=obj.get("tags"),
        write_access=obj.get("writeAccess"),
        type=obj.get("type"),
        raw=obj,
        **kwargs,
    )

get_by_id async classmethod 

get_by_id(
    auth: DomoAuth,
    objective_id: int | str,
    period_id: int | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoObjective

Retrieve a single objective by ID.

The Goals API does not expose a dedicated "get objective by id" endpoint, so this searches all objectives and matches by ID.

Source code in src/crew_dcs/classes/DomoObjective/core.py 
 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
126
127
128
129
130
131
132
133
@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    objective_id: int | str,
    period_id: int | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoObjective:
    """Retrieve a single objective by ID.

    The Goals API does not expose a dedicated "get objective by id"
    endpoint, so this searches all objectives and matches by ID.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.get_objectives(
        auth=auth,
        period_id=period_id,
        context=context,
    )

    if return_raw:
        return res

    for obj in res.response:
        if str(obj.get("id")) == str(objective_id):
            return cls.from_dict(auth=auth, obj=obj)

    raise DomoObjective_NoMatchError(
        objective_id=objective_id, domo_instance=auth.domo_instance
    )

get_entity_by_id async classmethod 

get_entity_by_id(entity_id, **kwargs) -> DomoObjective

Alias for get_by_id.

Source code in src/crew_dcs/classes/DomoObjective/core.py 
135
136
137
138
@classmethod
async def get_entity_by_id(cls, entity_id, **kwargs) -> DomoObjective:
    """Alias for get_by_id."""
    return await cls.get_by_id(objective_id=entity_id, **kwargs)

update async 

update(
    payload: dict | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoObjective

Update this objective.

Source code in src/crew_dcs/classes/DomoObjective/core.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
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
async def update(
    self,
    payload: dict | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoObjective:
    """Update this objective."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    payload = payload or objectives_routes.generate_create_objective_body(
        name=self.name,
        period_id=self.period_id,
        description=self.description,
        status=self.status,
        owner_id=self.owner_id,
        owners=self.owners,
        assignees=self.assignees,
        parent_id=self.parent_id,
        tags=self.tags,
        write_access=self.write_access,
        type=self.type,
    )

    res = await objectives_routes.update_objective(
        auth=self.auth,
        objective_id=self.id,
        payload=payload,
        period_id=self.period_id,
        context=context,
    )

    if return_raw:
        return res

    updated = self.from_dict(auth=self.auth, obj=res.response)
    self.__dict__.update(
        {
            k: v
            for k, v in updated.__dict__.items()
            if k not in ("KeyResults", "Tags")
        }
    )
    return self

DomoObjectiveTag dataclass 

DomoObjectiveTag(
    id: int,
    name: str = None,
    category: dict[str, Any] = None,
    raw: dict = None,
)

Represents a single objective tag.

DomoObjectiveTags dataclass 

DomoObjectiveTags(
    parent: DomoEntity,
    tags: list[DomoObjectiveTag] = list(),
)

Bases: DomoSubEntity

Sub-entity for managing an objective's assigned tags.

create async staticmethod 

create(
    auth,
    name: str,
    category_id: int | None = None,
    category_name: str | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoObjectiveTag

Create a new objective tag.

Source code in src/crew_dcs/classes/DomoObjective/tag.py 
 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
@staticmethod
async def create(
    auth,
    name: str,
    category_id: int | None = None,
    category_name: str | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoObjectiveTag:
    """Create a new objective tag."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.create_tag(
        auth=auth,
        name=name,
        category_id=category_id,
        category_name=category_name,
        context=context,
    )

    if return_raw:
        return res

    return DomoObjectiveTag.from_dict(res.response)

delete async staticmethod 

delete(
    auth,
    tag_id: int | str,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Delete an objective tag.

Source code in src/crew_dcs/classes/DomoObjective/tag.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@staticmethod
async def delete(
    auth,
    tag_id: int | str,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Delete an objective tag."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await objectives_routes.delete_tag(
        auth=auth,
        tag_id=tag_id,
        context=context,
    )

get async 

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

Populate tags from the parent objective's raw data.

Source code in src/crew_dcs/classes/DomoObjective/tag.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def get(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoObjectiveTag]:
    """Populate tags from the parent objective's raw data."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    raw = getattr(self.parent, "raw", None) or {}
    self.tags = [DomoObjectiveTag.from_dict(obj) for obj in raw.get("tags", [])]
    return self.tags

list_all async staticmethod 

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

List every tag available in the Goals product (not just this objective's).

Source code in src/crew_dcs/classes/DomoObjective/tag.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@staticmethod
async def list_all(
    auth,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoObjectiveTag]:
    """List every tag available in the Goals product (not just this objective's)."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.get_tags(auth=auth, context=context)
    return [DomoObjectiveTag.from_dict(obj) for obj in res.response]

update async staticmethod 

update(
    auth,
    tag_id: int | str,
    payload: dict,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoObjectiveTag

Update an existing objective tag.

Source code in src/crew_dcs/classes/DomoObjective/tag.py 
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
@staticmethod
async def update(
    auth,
    tag_id: int | str,
    payload: dict,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoObjectiveTag:
    """Update an existing objective tag."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.update_tag(
        auth=auth,
        tag_id=tag_id,
        payload=payload,
        context=context,
    )

    return DomoObjectiveTag.from_dict(res.response)

DomoObjective_NoMatchError 

DomoObjective_NoMatchError(
    objective_id, domo_instance, **kwargs
)

Bases: ClassError

Raised when a specific objective cannot be found.

Source code in src/crew_dcs/classes/DomoObjective/core.py 
25
26
27
28
29
30
31
def __init__(self, objective_id, domo_instance, **kwargs):
    super().__init__(
        message=f"Objective '{objective_id}' not found",
        entity_id=objective_id,
        domo_instance=domo_instance,
        **kwargs,
    )

DomoObjectives dataclass 

DomoObjectives(
    auth: DomoAuth, objectives: list[DomoObjective] = None
)

Bases: DomoManager

Manager for retrieving collections of objectives.

get async 

get(
    period_id: int | None = None,
    filter_key_results: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoObjective]

Search for objectives.

Source code in src/crew_dcs/classes/DomoObjective/core.py 
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
async def get(
    self,
    period_id: int | None = None,
    filter_key_results: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoObjective]:
    """Search for objectives."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await objectives_routes.get_objectives(
        auth=self.auth,
        period_id=period_id,
        filter_key_results=filter_key_results,
        context=context,
    )

    if return_raw:
        return res

    self.objectives = [
        DomoObjective.from_dict(auth=self.auth, obj=obj) for obj in res.response
    ]
    return self.objectives

Modules