Skip to content

metrics

metrics

get_metrics async 

get_metrics(
    auth: DomoAuth,
    filter_body: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Retrieve metrics via the filter endpoint.

Passes filter_body verbatim when provided; uses sensible defaults otherwise.

Source code in src/crew_dcs/routes/metrics.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
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_metrics(
    auth: DomoAuth,
    filter_body: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Retrieve metrics via the filter endpoint.

    Passes filter_body verbatim when provided; uses sensible defaults otherwise.
    """
    url = f"https://{auth.domo_instance}.domo.com/api/content/v1/metrics/filter"
    body = filter_body if filter_body is not None else dict(_DEFAULT_FILTER_BODY)

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Metric_GET_Error(res=res)

    return cast(rgd.ResponseGetData, res)

get_owned_metrics async 

get_owned_metrics(
    auth: DomoAuth,
    user_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Retrieve metrics owned by a specific user.

Source code in src/crew_dcs/routes/metrics.py 
 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
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_owned_metrics(
    auth: DomoAuth,
    user_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Retrieve metrics owned by a specific user."""
    url = f"https://{auth.domo_instance}.domo.com/api/content/v1/metrics/filter"
    body = {
        **_DEFAULT_FILTER_BODY,
        "filters": {"OWNER": [user_id]},
    }

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Metric_GET_Error(res=res)

    return cast(rgd.ResponseGetData, res)

transfer_metrics async 

transfer_metrics(
    auth: DomoAuth,
    from_user_id: str,
    to_user_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Transfer all metrics owned by from_user_id to to_user_id.

Source code in src/crew_dcs/routes/metrics.py 
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
194
195
196
197
198
199
200
201
202
203
204
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def transfer_metrics(
    auth: DomoAuth,
    from_user_id: str,
    to_user_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Transfer all metrics owned by from_user_id to to_user_id."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    owned_res = await get_owned_metrics(
        auth=auth,
        user_id=from_user_id,
        context=context,
    )

    metrics = owned_res.response if isinstance(owned_res.response, list) else []

    for metric in metrics:
        metric_id = str(metric.get("id") or metric.get("metricId") or "")
        if not metric_id:
            continue
        await update_metric_owner(
            auth=auth,
            metric_id=metric_id,
            to_user_id=to_user_id,
            context=context,
        )

    owned_res.response = {
        "transferred": len(metrics),
        "from_user_id": from_user_id,
        "to_user_id": to_user_id,
    }

    return cast(rgd.ResponseGetData, owned_res)

update_metric_owner async 

update_metric_owner(
    auth: DomoAuth,
    metric_id: str,
    to_user_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Transfer ownership of a metric to a different user.

Note: the domo-toolkit source uses POST (not PUT) for this endpoint.

Source code in src/crew_dcs/routes/metrics.py 
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
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_metric_owner(
    auth: DomoAuth,
    metric_id: str,
    to_user_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Transfer ownership of a metric to a different user.

    Note: the domo-toolkit source uses POST (not PUT) for this endpoint.
    """
    url = f"https://{auth.domo_instance}.domo.com/api/content/v1/metrics/{metric_id}/owner/{to_user_id}"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        context=context,
    )

    if not res.is_success:
        raise Metric_CRUD_Error(
            res=res,
            message=f"Failed to transfer metric {metric_id} to user {to_user_id}",
        )

    return cast(rgd.ResponseGetData, res)