Skip to content

datarobot_genai.drtools.core.clients.datarobot_workload

datarobot_workload

WorkloadApiClient

Workload API methods backed by the per-request DataRobot REST client.

Each call runs inside :func:request_user_dr_client so credentials come from the requesting user's headers::

client = WorkloadApiClient()
result = client.list_workloads(limit=50)
Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
 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
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
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
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
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
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
class WorkloadApiClient:
    """Workload API methods backed by the per-request DataRobot REST client.

    Each call runs inside :func:`request_user_dr_client` so credentials come from
    the requesting user's headers::

        client = WorkloadApiClient()
        result = client.list_workloads(limit=50)
    """

    # ------------------------------------------------------------------ #
    # Workloads — read                                                     #
    # ------------------------------------------------------------------ #

    def list_workloads(
        self,
        *,
        limit: int = 100,
        offset: int = 0,
        search: str | None = None,
    ) -> dict[str, Any]:
        """GET /workloads/ with optional server-side search and pagination."""
        params: dict[str, Any] = {"limit": limit, "offset": offset}
        if search:
            params["search"] = search
        with request_user_dr_client() as client:
            return client.get("workloads/", params=params).json()

    def get_workload(self, workload_id: str) -> dict[str, Any]:
        """GET /workloads/{workload_id}."""
        with request_user_dr_client() as client:
            return client.get(f"workloads/{workload_id}").json()

    # ------------------------------------------------------------------ #
    # Compute bundles                                                      #
    # ------------------------------------------------------------------ #

    def list_bundles(self) -> dict[str, Any]:
        """GET /mlops/compute/bundles — available CPU/GPU resource bundles."""
        with request_user_dr_client() as client:
            return client.get("mlops/compute/bundles").json()

    # ------------------------------------------------------------------ #
    # Workloads — write                                                    #
    # ------------------------------------------------------------------ #

    def create_workload(self, payload: dict[str, Any]) -> dict[str, Any]:
        """POST /workloads/ — returns the created WorkloadFormatted (201)."""
        with request_user_dr_client() as client:
            return client.post("workloads/", json=payload).json()

    def start_workload(self, workload_id: str) -> dict[str, Any]:
        """POST /workloads/{id}/start — returns WorkloadOperationResponse (202)."""
        with request_user_dr_client() as client:
            resp = client.post(f"workloads/{workload_id}/start")
            return resp.json() if resp.content else {}

    def stop_workload(self, workload_id: str) -> dict[str, Any]:
        """POST /workloads/{id}/stop — returns WorkloadOperationResponse (202)."""
        with request_user_dr_client() as client:
            resp = client.post(f"workloads/{workload_id}/stop")
            return resp.json() if resp.content else {}

    def delete_workload(self, workload_id: str) -> None:
        """DELETE /workloads/{id} — 204 No Content on success."""
        with request_user_dr_client() as client:
            client.delete(f"workloads/{workload_id}")

    def patch_workload(self, workload_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        """PATCH /workloads/{id} — partial update of name / description / importance."""
        with request_user_dr_client() as client:
            return client.patch(f"workloads/{workload_id}", json=payload).json()

    # ------------------------------------------------------------------ #
    # Workloads — settings                                                 #
    # ------------------------------------------------------------------ #

    def get_workload_settings(self, workload_id: str) -> dict[str, Any]:
        """GET /workloads/{id}/settings — returns WorkloadSettingsResponse."""
        with request_user_dr_client() as client:
            return client.get(f"workloads/{workload_id}/settings").json()

    def update_workload_settings(self, workload_id: str, runtime: dict[str, Any]) -> dict[str, Any]:
        """PATCH /workloads/{id}/settings — triggers rolling replacement (202)."""
        with request_user_dr_client() as client:
            return client.patch(
                f"workloads/{workload_id}/settings", json={"runtime": runtime}
            ).json()

    # ------------------------------------------------------------------ #
    # Workloads — observability                                            #
    # ------------------------------------------------------------------ #

    def get_workload_stats(
        self,
        workload_id: str,
        *,
        proton_id: str | None = None,
        start_time: str | None = None,
        end_time: str | None = None,
        response_time_quantile: float = 0.5,
        slow_requests_threshold: int = 2000,
    ) -> dict[str, Any]:
        """GET /workloads/{id}/stats — aggregated performance statistics."""
        params: dict[str, Any] = {
            "responseTimeQuantile": response_time_quantile,
            "slowRequestsThreshold": slow_requests_threshold,
        }
        if proton_id:
            params["protonId"] = proton_id
        if start_time:
            params["startTime"] = start_time
        if end_time:
            params["endTime"] = end_time
        with request_user_dr_client() as client:
            return client.get(f"workloads/{workload_id}/stats", params=params).json()

    def list_workload_history(
        self, workload_id: str, *, limit: int = 10, offset: int = 0
    ) -> dict[str, Any]:
        """GET /workloads/{id}/history — artifact deployment history."""
        with request_user_dr_client() as client:
            return client.get(
                f"workloads/{workload_id}/history",
                params={"limit": limit, "offset": offset},
            ).json()

    def list_workload_events(
        self, workload_id: str, *, limit: int = 10, offset: int = 0
    ) -> dict[str, Any]:
        """GET /workloads/{id}/events — status-change and error events."""
        with request_user_dr_client() as client:
            return client.get(
                f"workloads/{workload_id}/events",
                params={"limit": limit, "offset": offset},
            ).json()

    def promote_workload_artifact(self, workload_id: str) -> dict[str, Any]:
        """POST /workloads/{id}/promote — lock the running draft artifact (202)."""
        with request_user_dr_client() as client:
            resp = client.post(f"workloads/{workload_id}/promote")
            return resp.json() if resp.content else {}

    def get_workload_related(self, workload_id: str) -> dict[str, Any]:
        """GET /workloads/{id}/related — linked artifacts and related entities."""
        with request_user_dr_client() as client:
            return client.get(f"workloads/{workload_id}/related").json()

    # ------------------------------------------------------------------ #
    # Protons                                                              #
    # ------------------------------------------------------------------ #

    def list_protons(self, workload_id: str, *, limit: int = 10, offset: int = 0) -> dict[str, Any]:
        """GET /workloads/{id}/protons — deployed proton instances for a workload."""
        with request_user_dr_client() as client:
            return client.get(
                f"workloads/{workload_id}/protons",
                params={"limit": limit, "offset": offset},
            ).json()

    def get_proton(self, workload_id: str, proton_id: str) -> dict[str, Any]:
        """GET /workloads/{id}/protons/{proton_id} — single proton record."""
        with request_user_dr_client() as client:
            return client.get(f"workloads/{workload_id}/protons/{proton_id}").json()

    def get_proton_status_details(self, workload_id: str, proton_id: str) -> dict[str, Any] | None:
        """GET /workloads/{id}/protons/{proton_id}/statusDetails.

        Returns the ReplicaStatusesSnapshot when available (200), or
        ``None`` when no status has been received yet (204).
        """
        with request_user_dr_client() as client:
            resp = client.get(f"workloads/{workload_id}/protons/{proton_id}/statusDetails")
            return resp.json() if resp.content else None

    # ------------------------------------------------------------------ #
    # OTel logs                                                            #
    # ------------------------------------------------------------------ #

    def list_workload_logs(
        self,
        workload_id: str,
        *,
        limit: int = 100,
        offset: int = 0,
        level: str = "debug",
        start_time: str | None = None,
        end_time: str | None = None,
        includes: list[str] | None = None,
        excludes: list[str] | None = None,
        span_id: str | None = None,
        trace_id: str | None = None,
    ) -> dict[str, Any]:
        """GET /otel/workload/{id}/logs/ — OTel log lines for a workload."""
        params: dict[str, Any] = {"limit": limit, "offset": offset, "level": level}
        if start_time:
            params["startTime"] = start_time
        if end_time:
            params["endTime"] = end_time
        if includes:
            params["includes"] = includes
        if excludes:
            params["excludes"] = excludes
        if span_id:
            params["spanId"] = span_id
        if trace_id:
            params["traceId"] = trace_id
        with request_user_dr_client() as client:
            return client.get(f"otel/workload/{workload_id}/logs/", params=params).json()

    # ------------------------------------------------------------------ #
    # Artifacts                                                            #
    # ------------------------------------------------------------------ #

    def list_artifacts(
        self,
        *,
        limit: int = 100,
        offset: int = 0,
        search: str | None = None,
        status: str | None = None,
        artifact_type: str | None = None,
        repository_id: str | None = None,
    ) -> dict[str, Any]:
        """GET /artifacts/ with optional filters and pagination."""
        params: dict[str, Any] = {"limit": limit, "offset": offset}
        if search:
            params["search"] = search
        if status:
            params["status"] = status
        if artifact_type:
            params["type"] = artifact_type
        if repository_id:
            params["repositoryId"] = repository_id
        with request_user_dr_client() as client:
            return client.get("artifacts/", params=params).json()

    def get_artifact(self, artifact_id: str) -> dict[str, Any]:
        """GET /artifacts/{id}."""
        with request_user_dr_client() as client:
            return client.get(f"artifacts/{artifact_id}").json()

    def create_artifact(self, payload: dict[str, Any]) -> dict[str, Any]:
        """POST /artifacts/ — returns the created ArtifactFormatted (201)."""
        with request_user_dr_client() as client:
            return client.post("artifacts/", json=payload).json()

    def put_artifact(self, artifact_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        """PUT /artifacts/{id} — full replacement with InputArtifact payload."""
        with request_user_dr_client() as client:
            return client.put(f"artifacts/{artifact_id}", json=payload).json()

    def patch_artifact(self, artifact_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        """PATCH /artifacts/{id} — partial update via UpdateArtifactRequest."""
        with request_user_dr_client() as client:
            return client.patch(f"artifacts/{artifact_id}", json=payload).json()

    def delete_artifact(self, artifact_id: str) -> None:
        """DELETE /artifacts/{id} — 204 No Content on success."""
        with request_user_dr_client() as client:
            client.delete(f"artifacts/{artifact_id}")

    def clone_artifact(self, artifact_id: str, name: str) -> dict[str, Any]:
        """POST /artifacts/{id}/clone — duplicate with a new name."""
        with request_user_dr_client() as client:
            return client.post(f"artifacts/{artifact_id}/clone", json={"name": name}).json()

    # ------------------------------------------------------------------ #
    # Artifact builds                                                      #
    # ------------------------------------------------------------------ #

    def list_artifact_builds(
        self,
        artifact_id: str,
        *,
        limit: int = 100,
        offset: int = 0,
    ) -> dict[str, Any]:
        """GET /artifacts/{id}/builds."""
        params: dict[str, Any] = {"limit": limit, "offset": offset}
        with request_user_dr_client() as client:
            return client.get(f"artifacts/{artifact_id}/builds", params=params).json()

    def trigger_artifact_build(self, artifact_id: str) -> dict[str, Any]:
        """POST /artifacts/{id}/builds — start image build(s) for a draft service artifact."""
        with request_user_dr_client() as client:
            return client.post(f"artifacts/{artifact_id}/builds").json()

    def get_artifact_build(self, artifact_id: str, build_id: str) -> dict[str, Any]:
        """GET /artifacts/{id}/builds/{build_id}."""
        with request_user_dr_client() as client:
            return client.get(f"artifacts/{artifact_id}/builds/{build_id}").json()

    def get_artifact_build_logs(self, artifact_id: str, build_id: str) -> str:
        """GET /artifacts/{id}/builds/{build_id}/logs — returns raw log text."""
        with request_user_dr_client() as client:
            return client.get(f"artifacts/{artifact_id}/builds/{build_id}/logs").text

    def delete_artifact_build(self, artifact_id: str, build_id: str) -> None:
        """DELETE /artifacts/{id}/builds/{build_id} — cancel or delete a build (204)."""
        with request_user_dr_client() as client:
            client.delete(f"artifacts/{artifact_id}/builds/{build_id}")

    # ------------------------------------------------------------------ #
    # Artifact repositories                                                #
    # ------------------------------------------------------------------ #

    def list_artifact_repositories(
        self,
        *,
        limit: int = 100,
        offset: int = 0,
        search: str | None = None,
        artifact_type: str | None = None,
    ) -> dict[str, Any]:
        """GET /artifactRepositories."""
        params: dict[str, Any] = {"limit": limit, "offset": offset}
        if search:
            params["search"] = search
        if artifact_type:
            params["type"] = artifact_type
        with request_user_dr_client() as client:
            return client.get("artifactRepositories", params=params).json()

    def get_artifact_repository(self, repository_id: str) -> dict[str, Any]:
        """GET /artifactRepositories/{id}."""
        with request_user_dr_client() as client:
            return client.get(f"artifactRepositories/{repository_id}").json()

    def delete_artifact_repository(self, repository_id: str) -> None:
        """DELETE /artifactRepositories/{id} — 204 No Content on success."""
        with request_user_dr_client() as client:
            client.delete(f"artifactRepositories/{repository_id}")

    # ------------------------------------------------------------------ #
    # Workload replacement (rolling update)                                #
    # ------------------------------------------------------------------ #

    def get_workload_replacement(self, workload_id: str) -> dict[str, Any]:
        """GET /workloads/{id}/replacement — current replacement status."""
        with request_user_dr_client() as client:
            return client.get(f"workloads/{workload_id}/replacement").json()

    def create_workload_replacement(
        self, workload_id: str, payload: dict[str, Any]
    ) -> dict[str, Any]:
        """POST /workloads/{id}/replacement — start a rolling replacement (202)."""
        with request_user_dr_client() as client:
            resp = client.post(f"workloads/{workload_id}/replacement", json=payload)
            return resp.json() if resp.content else {}

    def delete_workload_replacement(self, workload_id: str) -> dict[str, Any]:
        """DELETE /workloads/{id}/replacement — cancel an in-progress replacement (202)."""
        with request_user_dr_client() as client:
            resp = client.delete(f"workloads/{workload_id}/replacement")
            return resp.json() if resp.content else {}

list_workloads 

list_workloads(*, limit: int = 100, offset: int = 0, search: str | None = None) -> dict[str, Any]

GET /workloads/ with optional server-side search and pagination.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
def list_workloads(
    self,
    *,
    limit: int = 100,
    offset: int = 0,
    search: str | None = None,
) -> dict[str, Any]:
    """GET /workloads/ with optional server-side search and pagination."""
    params: dict[str, Any] = {"limit": limit, "offset": offset}
    if search:
        params["search"] = search
    with request_user_dr_client() as client:
        return client.get("workloads/", params=params).json()

get_workload 

get_workload(workload_id: str) -> dict[str, Any]

GET /workloads/{workload_id}.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
51
52
53
54
def get_workload(self, workload_id: str) -> dict[str, Any]:
    """GET /workloads/{workload_id}."""
    with request_user_dr_client() as client:
        return client.get(f"workloads/{workload_id}").json()

list_bundles 

list_bundles() -> dict[str, Any]

GET /mlops/compute/bundles — available CPU/GPU resource bundles.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
60
61
62
63
def list_bundles(self) -> dict[str, Any]:
    """GET /mlops/compute/bundles — available CPU/GPU resource bundles."""
    with request_user_dr_client() as client:
        return client.get("mlops/compute/bundles").json()

create_workload 

create_workload(payload: dict[str, Any]) -> dict[str, Any]

POST /workloads/ — returns the created WorkloadFormatted (201).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
69
70
71
72
def create_workload(self, payload: dict[str, Any]) -> dict[str, Any]:
    """POST /workloads/ — returns the created WorkloadFormatted (201)."""
    with request_user_dr_client() as client:
        return client.post("workloads/", json=payload).json()

start_workload 

start_workload(workload_id: str) -> dict[str, Any]

POST /workloads/{id}/start — returns WorkloadOperationResponse (202).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
74
75
76
77
78
def start_workload(self, workload_id: str) -> dict[str, Any]:
    """POST /workloads/{id}/start — returns WorkloadOperationResponse (202)."""
    with request_user_dr_client() as client:
        resp = client.post(f"workloads/{workload_id}/start")
        return resp.json() if resp.content else {}

stop_workload 

stop_workload(workload_id: str) -> dict[str, Any]

POST /workloads/{id}/stop — returns WorkloadOperationResponse (202).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
80
81
82
83
84
def stop_workload(self, workload_id: str) -> dict[str, Any]:
    """POST /workloads/{id}/stop — returns WorkloadOperationResponse (202)."""
    with request_user_dr_client() as client:
        resp = client.post(f"workloads/{workload_id}/stop")
        return resp.json() if resp.content else {}

delete_workload 

delete_workload(workload_id: str) -> None

DELETE /workloads/{id} — 204 No Content on success.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
86
87
88
89
def delete_workload(self, workload_id: str) -> None:
    """DELETE /workloads/{id} — 204 No Content on success."""
    with request_user_dr_client() as client:
        client.delete(f"workloads/{workload_id}")

patch_workload 

patch_workload(workload_id: str, payload: dict[str, Any]) -> dict[str, Any]

PATCH /workloads/{id} — partial update of name / description / importance.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
91
92
93
94
def patch_workload(self, workload_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    """PATCH /workloads/{id} — partial update of name / description / importance."""
    with request_user_dr_client() as client:
        return client.patch(f"workloads/{workload_id}", json=payload).json()

get_workload_settings 

get_workload_settings(workload_id: str) -> dict[str, Any]

GET /workloads/{id}/settings — returns WorkloadSettingsResponse.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
100
101
102
103
def get_workload_settings(self, workload_id: str) -> dict[str, Any]:
    """GET /workloads/{id}/settings — returns WorkloadSettingsResponse."""
    with request_user_dr_client() as client:
        return client.get(f"workloads/{workload_id}/settings").json()

update_workload_settings 

update_workload_settings(workload_id: str, runtime: dict[str, Any]) -> dict[str, Any]

PATCH /workloads/{id}/settings — triggers rolling replacement (202).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
105
106
107
108
109
110
def update_workload_settings(self, workload_id: str, runtime: dict[str, Any]) -> dict[str, Any]:
    """PATCH /workloads/{id}/settings — triggers rolling replacement (202)."""
    with request_user_dr_client() as client:
        return client.patch(
            f"workloads/{workload_id}/settings", json={"runtime": runtime}
        ).json()

get_workload_stats 

get_workload_stats(workload_id: str, *, proton_id: str | None = None, start_time: str | None = None, end_time: str | None = None, response_time_quantile: float = 0.5, slow_requests_threshold: int = 2000) -> dict[str, Any]

GET /workloads/{id}/stats — aggregated performance statistics.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def get_workload_stats(
    self,
    workload_id: str,
    *,
    proton_id: str | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    response_time_quantile: float = 0.5,
    slow_requests_threshold: int = 2000,
) -> dict[str, Any]:
    """GET /workloads/{id}/stats — aggregated performance statistics."""
    params: dict[str, Any] = {
        "responseTimeQuantile": response_time_quantile,
        "slowRequestsThreshold": slow_requests_threshold,
    }
    if proton_id:
        params["protonId"] = proton_id
    if start_time:
        params["startTime"] = start_time
    if end_time:
        params["endTime"] = end_time
    with request_user_dr_client() as client:
        return client.get(f"workloads/{workload_id}/stats", params=params).json()

list_workload_history 

list_workload_history(workload_id: str, *, limit: int = 10, offset: int = 0) -> dict[str, Any]

GET /workloads/{id}/history — artifact deployment history.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
140
141
142
143
144
145
146
147
148
def list_workload_history(
    self, workload_id: str, *, limit: int = 10, offset: int = 0
) -> dict[str, Any]:
    """GET /workloads/{id}/history — artifact deployment history."""
    with request_user_dr_client() as client:
        return client.get(
            f"workloads/{workload_id}/history",
            params={"limit": limit, "offset": offset},
        ).json()

list_workload_events 

list_workload_events(workload_id: str, *, limit: int = 10, offset: int = 0) -> dict[str, Any]

GET /workloads/{id}/events — status-change and error events.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
150
151
152
153
154
155
156
157
158
def list_workload_events(
    self, workload_id: str, *, limit: int = 10, offset: int = 0
) -> dict[str, Any]:
    """GET /workloads/{id}/events — status-change and error events."""
    with request_user_dr_client() as client:
        return client.get(
            f"workloads/{workload_id}/events",
            params={"limit": limit, "offset": offset},
        ).json()

promote_workload_artifact 

promote_workload_artifact(workload_id: str) -> dict[str, Any]

POST /workloads/{id}/promote — lock the running draft artifact (202).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
160
161
162
163
164
def promote_workload_artifact(self, workload_id: str) -> dict[str, Any]:
    """POST /workloads/{id}/promote — lock the running draft artifact (202)."""
    with request_user_dr_client() as client:
        resp = client.post(f"workloads/{workload_id}/promote")
        return resp.json() if resp.content else {}

get_workload_related(workload_id: str) -> dict[str, Any]

GET /workloads/{id}/related — linked artifacts and related entities.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
166
167
168
169
def get_workload_related(self, workload_id: str) -> dict[str, Any]:
    """GET /workloads/{id}/related — linked artifacts and related entities."""
    with request_user_dr_client() as client:
        return client.get(f"workloads/{workload_id}/related").json()

list_protons 

list_protons(workload_id: str, *, limit: int = 10, offset: int = 0) -> dict[str, Any]

GET /workloads/{id}/protons — deployed proton instances for a workload.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
175
176
177
178
179
180
181
def list_protons(self, workload_id: str, *, limit: int = 10, offset: int = 0) -> dict[str, Any]:
    """GET /workloads/{id}/protons — deployed proton instances for a workload."""
    with request_user_dr_client() as client:
        return client.get(
            f"workloads/{workload_id}/protons",
            params={"limit": limit, "offset": offset},
        ).json()

get_proton 

get_proton(workload_id: str, proton_id: str) -> dict[str, Any]

GET /workloads/{id}/protons/{proton_id} — single proton record.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
183
184
185
186
def get_proton(self, workload_id: str, proton_id: str) -> dict[str, Any]:
    """GET /workloads/{id}/protons/{proton_id} — single proton record."""
    with request_user_dr_client() as client:
        return client.get(f"workloads/{workload_id}/protons/{proton_id}").json()

get_proton_status_details 

get_proton_status_details(workload_id: str, proton_id: str) -> dict[str, Any] | None

GET /workloads/{id}/protons/{proton_id}/statusDetails.

Returns the ReplicaStatusesSnapshot when available (200), or None when no status has been received yet (204).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
188
189
190
191
192
193
194
195
196
def get_proton_status_details(self, workload_id: str, proton_id: str) -> dict[str, Any] | None:
    """GET /workloads/{id}/protons/{proton_id}/statusDetails.

    Returns the ReplicaStatusesSnapshot when available (200), or
    ``None`` when no status has been received yet (204).
    """
    with request_user_dr_client() as client:
        resp = client.get(f"workloads/{workload_id}/protons/{proton_id}/statusDetails")
        return resp.json() if resp.content else None

list_workload_logs 

list_workload_logs(workload_id: str, *, limit: int = 100, offset: int = 0, level: str = 'debug', start_time: str | None = None, end_time: str | None = None, includes: list[str] | None = None, excludes: list[str] | None = None, span_id: str | None = None, trace_id: str | None = None) -> dict[str, Any]

GET /otel/workload/{id}/logs/ — OTel log lines for a workload.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
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
def list_workload_logs(
    self,
    workload_id: str,
    *,
    limit: int = 100,
    offset: int = 0,
    level: str = "debug",
    start_time: str | None = None,
    end_time: str | None = None,
    includes: list[str] | None = None,
    excludes: list[str] | None = None,
    span_id: str | None = None,
    trace_id: str | None = None,
) -> dict[str, Any]:
    """GET /otel/workload/{id}/logs/ — OTel log lines for a workload."""
    params: dict[str, Any] = {"limit": limit, "offset": offset, "level": level}
    if start_time:
        params["startTime"] = start_time
    if end_time:
        params["endTime"] = end_time
    if includes:
        params["includes"] = includes
    if excludes:
        params["excludes"] = excludes
    if span_id:
        params["spanId"] = span_id
    if trace_id:
        params["traceId"] = trace_id
    with request_user_dr_client() as client:
        return client.get(f"otel/workload/{workload_id}/logs/", params=params).json()

list_artifacts 

list_artifacts(*, limit: int = 100, offset: int = 0, search: str | None = None, status: str | None = None, artifact_type: str | None = None, repository_id: str | None = None) -> dict[str, Any]

GET /artifacts/ with optional filters and pagination.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def list_artifacts(
    self,
    *,
    limit: int = 100,
    offset: int = 0,
    search: str | None = None,
    status: str | None = None,
    artifact_type: str | None = None,
    repository_id: str | None = None,
) -> dict[str, Any]:
    """GET /artifacts/ with optional filters and pagination."""
    params: dict[str, Any] = {"limit": limit, "offset": offset}
    if search:
        params["search"] = search
    if status:
        params["status"] = status
    if artifact_type:
        params["type"] = artifact_type
    if repository_id:
        params["repositoryId"] = repository_id
    with request_user_dr_client() as client:
        return client.get("artifacts/", params=params).json()

get_artifact 

get_artifact(artifact_id: str) -> dict[str, Any]

GET /artifacts/{id}.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
260
261
262
263
def get_artifact(self, artifact_id: str) -> dict[str, Any]:
    """GET /artifacts/{id}."""
    with request_user_dr_client() as client:
        return client.get(f"artifacts/{artifact_id}").json()

create_artifact 

create_artifact(payload: dict[str, Any]) -> dict[str, Any]

POST /artifacts/ — returns the created ArtifactFormatted (201).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
265
266
267
268
def create_artifact(self, payload: dict[str, Any]) -> dict[str, Any]:
    """POST /artifacts/ — returns the created ArtifactFormatted (201)."""
    with request_user_dr_client() as client:
        return client.post("artifacts/", json=payload).json()

put_artifact 

put_artifact(artifact_id: str, payload: dict[str, Any]) -> dict[str, Any]

PUT /artifacts/{id} — full replacement with InputArtifact payload.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
270
271
272
273
def put_artifact(self, artifact_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    """PUT /artifacts/{id} — full replacement with InputArtifact payload."""
    with request_user_dr_client() as client:
        return client.put(f"artifacts/{artifact_id}", json=payload).json()

patch_artifact 

patch_artifact(artifact_id: str, payload: dict[str, Any]) -> dict[str, Any]

PATCH /artifacts/{id} — partial update via UpdateArtifactRequest.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
275
276
277
278
def patch_artifact(self, artifact_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    """PATCH /artifacts/{id} — partial update via UpdateArtifactRequest."""
    with request_user_dr_client() as client:
        return client.patch(f"artifacts/{artifact_id}", json=payload).json()

delete_artifact 

delete_artifact(artifact_id: str) -> None

DELETE /artifacts/{id} — 204 No Content on success.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
280
281
282
283
def delete_artifact(self, artifact_id: str) -> None:
    """DELETE /artifacts/{id} — 204 No Content on success."""
    with request_user_dr_client() as client:
        client.delete(f"artifacts/{artifact_id}")

clone_artifact 

clone_artifact(artifact_id: str, name: str) -> dict[str, Any]

POST /artifacts/{id}/clone — duplicate with a new name.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
285
286
287
288
def clone_artifact(self, artifact_id: str, name: str) -> dict[str, Any]:
    """POST /artifacts/{id}/clone — duplicate with a new name."""
    with request_user_dr_client() as client:
        return client.post(f"artifacts/{artifact_id}/clone", json={"name": name}).json()

list_artifact_builds 

list_artifact_builds(artifact_id: str, *, limit: int = 100, offset: int = 0) -> dict[str, Any]

GET /artifacts/{id}/builds.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
294
295
296
297
298
299
300
301
302
303
304
def list_artifact_builds(
    self,
    artifact_id: str,
    *,
    limit: int = 100,
    offset: int = 0,
) -> dict[str, Any]:
    """GET /artifacts/{id}/builds."""
    params: dict[str, Any] = {"limit": limit, "offset": offset}
    with request_user_dr_client() as client:
        return client.get(f"artifacts/{artifact_id}/builds", params=params).json()

trigger_artifact_build 

trigger_artifact_build(artifact_id: str) -> dict[str, Any]

POST /artifacts/{id}/builds — start image build(s) for a draft service artifact.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
306
307
308
309
def trigger_artifact_build(self, artifact_id: str) -> dict[str, Any]:
    """POST /artifacts/{id}/builds — start image build(s) for a draft service artifact."""
    with request_user_dr_client() as client:
        return client.post(f"artifacts/{artifact_id}/builds").json()

get_artifact_build 

get_artifact_build(artifact_id: str, build_id: str) -> dict[str, Any]

GET /artifacts/{id}/builds/{build_id}.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
311
312
313
314
def get_artifact_build(self, artifact_id: str, build_id: str) -> dict[str, Any]:
    """GET /artifacts/{id}/builds/{build_id}."""
    with request_user_dr_client() as client:
        return client.get(f"artifacts/{artifact_id}/builds/{build_id}").json()

get_artifact_build_logs 

get_artifact_build_logs(artifact_id: str, build_id: str) -> str

GET /artifacts/{id}/builds/{build_id}/logs — returns raw log text.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
316
317
318
319
def get_artifact_build_logs(self, artifact_id: str, build_id: str) -> str:
    """GET /artifacts/{id}/builds/{build_id}/logs — returns raw log text."""
    with request_user_dr_client() as client:
        return client.get(f"artifacts/{artifact_id}/builds/{build_id}/logs").text

delete_artifact_build 

delete_artifact_build(artifact_id: str, build_id: str) -> None

DELETE /artifacts/{id}/builds/{build_id} — cancel or delete a build (204).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
321
322
323
324
def delete_artifact_build(self, artifact_id: str, build_id: str) -> None:
    """DELETE /artifacts/{id}/builds/{build_id} — cancel or delete a build (204)."""
    with request_user_dr_client() as client:
        client.delete(f"artifacts/{artifact_id}/builds/{build_id}")

list_artifact_repositories 

list_artifact_repositories(*, limit: int = 100, offset: int = 0, search: str | None = None, artifact_type: str | None = None) -> dict[str, Any]

GET /artifactRepositories.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def list_artifact_repositories(
    self,
    *,
    limit: int = 100,
    offset: int = 0,
    search: str | None = None,
    artifact_type: str | None = None,
) -> dict[str, Any]:
    """GET /artifactRepositories."""
    params: dict[str, Any] = {"limit": limit, "offset": offset}
    if search:
        params["search"] = search
    if artifact_type:
        params["type"] = artifact_type
    with request_user_dr_client() as client:
        return client.get("artifactRepositories", params=params).json()

get_artifact_repository 

get_artifact_repository(repository_id: str) -> dict[str, Any]

GET /artifactRepositories/{id}.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
347
348
349
350
def get_artifact_repository(self, repository_id: str) -> dict[str, Any]:
    """GET /artifactRepositories/{id}."""
    with request_user_dr_client() as client:
        return client.get(f"artifactRepositories/{repository_id}").json()

delete_artifact_repository 

delete_artifact_repository(repository_id: str) -> None

DELETE /artifactRepositories/{id} — 204 No Content on success.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
352
353
354
355
def delete_artifact_repository(self, repository_id: str) -> None:
    """DELETE /artifactRepositories/{id} — 204 No Content on success."""
    with request_user_dr_client() as client:
        client.delete(f"artifactRepositories/{repository_id}")

get_workload_replacement 

get_workload_replacement(workload_id: str) -> dict[str, Any]

GET /workloads/{id}/replacement — current replacement status.

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
361
362
363
364
def get_workload_replacement(self, workload_id: str) -> dict[str, Any]:
    """GET /workloads/{id}/replacement — current replacement status."""
    with request_user_dr_client() as client:
        return client.get(f"workloads/{workload_id}/replacement").json()

create_workload_replacement 

create_workload_replacement(workload_id: str, payload: dict[str, Any]) -> dict[str, Any]

POST /workloads/{id}/replacement — start a rolling replacement (202).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
366
367
368
369
370
371
372
def create_workload_replacement(
    self, workload_id: str, payload: dict[str, Any]
) -> dict[str, Any]:
    """POST /workloads/{id}/replacement — start a rolling replacement (202)."""
    with request_user_dr_client() as client:
        resp = client.post(f"workloads/{workload_id}/replacement", json=payload)
        return resp.json() if resp.content else {}

delete_workload_replacement 

delete_workload_replacement(workload_id: str) -> dict[str, Any]

DELETE /workloads/{id}/replacement — cancel an in-progress replacement (202).

Source code in datarobot_genai/drtools/core/clients/datarobot_workload.py 
374
375
376
377
378
def delete_workload_replacement(self, workload_id: str) -> dict[str, Any]:
    """DELETE /workloads/{id}/replacement — cancel an in-progress replacement (202)."""
    with request_user_dr_client() as client:
        resp = client.delete(f"workloads/{workload_id}/replacement")
        return resp.json() if resp.content else {}