from __future__ import annotations
import os
import re
import time
import requests as _requests
from .exceptions import (
AuthenticationError,
ConcurrencyLimitError,
ConnectionError,
DazBusyError,
ScriptRuntimeError,
ScriptSyntaxError,
StudioBusyError,
TimeoutError,
)
from ._result import ExecutionResult
_TOKEN_FILE = os.path.expanduser("~/.daz3d/dazscriptserver_token.txt")
_DEFAULT_HOST = "127.0.0.1"
_DEFAULT_PORT = 18811
def _load_token() -> str:
if os.path.exists(_TOKEN_FILE):
with open(_TOKEN_FILE) as f:
return f.read().strip()
return ""
def _parse_retry_after(resp: _requests.Response) -> float:
raw = resp.headers.get("Retry-After", "")
try:
return float(raw)
except (TypeError, ValueError):
return 2.0
def _raise_for_error(resp: _requests.Response) -> None:
"""Raise a typed exception for authentication or server-busy responses.
Leaves 2xx responses (and DazScript's own success:false runtime/syntax
errors, which use HTTP 200) for the caller to handle.
"""
status = resp.status_code
if status == 401 or status == 403:
raise AuthenticationError(f"HTTP {status}: {resp.text[:200]}")
if status < 400:
return
data = resp.json()
error_code = data.get("error_code", "")
error_msg = data.get("error") or f"HTTP {status}"
retry_after = _parse_retry_after(resp)
if error_code == "STUDIO_BUSY":
raise StudioBusyError(error_msg, reason=data.get("detail", error_msg), retry_after=retry_after)
if error_code == "CONCURRENT_LIMIT_EXCEEDED":
raise ConcurrencyLimitError(error_msg, reason=error_msg, retry_after=retry_after)
def _map_response(resp: _requests.Response, script: str = "") -> ExecutionResult:
_raise_for_error(resp)
data = resp.json()
request_id = data.get("request_id", "")
if not data.get("success", True):
error_msg = data.get("error", "Script failed")
output = data.get("output", [])
# SyntaxError comes from the parser; runtime errors (TypeError, ReferenceError,
# Error, etc.) also include "Line N:" but never say "SyntaxError" explicitly.
if "SyntaxError" in error_msg:
raise ScriptSyntaxError(error_msg, script=script, request_id=request_id, output=output)
raise ScriptRuntimeError(error_msg, script=script, request_id=request_id, output=output)
return ExecutionResult(
value=data.get("result"),
output=data.get("output", []),
request_id=request_id,
success=True,
error="",
duration_ms=data.get("duration_ms", 0.0),
)
[docs]
class DazClient:
"""HTTP client for the DAZ Studio Script Server.
Handles authentication, request serialisation, and response mapping for
all server endpoints. The token is loaded automatically from
``~/.daz3d/dazscriptserver_token.txt`` when *token* is ``None``.
Args:
host: Hostname or IP address of the Script Server.
port: Listening port of the Script Server.
token: API token. Pass an empty string to disable authentication or
``None`` to auto-load from the default token file.
timeout: Per-request HTTP timeout in seconds.
Requests are issued through a pooled ``requests.Session`` for connection
reuse. Call :meth:`close` when done, or use as a context manager::
client = DazClient() # default 127.0.0.1:18811
client = DazClient(token="my-secret-token") # explicit token
with DazClient() as client:
result = client.execute("1 + 1;")
"""
def __init__(
self,
host: str = _DEFAULT_HOST,
port: int = _DEFAULT_PORT,
token: str | None = None,
timeout: float = 30.0,
):
self._base = f"http://{host}:{port}"
self._token = token if token is not None else _load_token()
self._timeout = timeout
self._session = _requests.Session()
def __enter__(self) -> "DazClient":
return self
def __exit__(self, *exc_info) -> None:
self.close()
[docs]
def close(self) -> None:
"""Close the underlying ``requests.Session``'s pooled connections."""
self._session.close()
@property
def _headers(self) -> dict:
h = {}
if self._token:
h["X-API-Token"] = self._token
return h
def _post(self, path: str, payload: dict) -> _requests.Response:
try:
return self._session.post(
f"{self._base}{path}",
json=payload,
headers=self._headers,
timeout=self._timeout,
)
except _requests.exceptions.ConnectionError as e:
raise ConnectionError(f"Cannot reach DAZ Studio at {self._base}: {e}") from e
except _requests.exceptions.Timeout as e:
raise TimeoutError(f"Request timed out after {self._timeout}s") from e
def _get(self, path: str, params: dict | None = None) -> _requests.Response:
try:
return self._session.get(
f"{self._base}{path}",
headers=self._headers,
params=params,
timeout=self._timeout,
)
except _requests.exceptions.ConnectionError as e:
raise ConnectionError(f"Cannot reach DAZ Studio at {self._base}: {e}") from e
except _requests.exceptions.Timeout as e:
raise TimeoutError(f"Request timed out after {self._timeout}s") from e
def _with_busy_retry(self, fn, retry_on_busy: bool, max_wait: float):
"""Run *fn* (a zero-arg callable), retrying on DazBusyError while
*retry_on_busy* is true, up to *max_wait* seconds total."""
if not retry_on_busy:
return fn()
deadline = time.monotonic() + max_wait
backoff = 1.0
while True:
try:
return fn()
except DazBusyError:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise
time.sleep(min(backoff, remaining))
backoff = min(backoff + 1.0, 5.0)
[docs]
def execute(
self, script: str, args: object = None, *, retry_on_busy: bool = False, max_wait: float = 30.0
) -> ExecutionResult:
"""Execute a DazScript string synchronously.
Args:
script: DazScript source code to execute.
args: Optional value passed into the script as ``getArguments()[0]``.
Must be JSON-serialisable.
retry_on_busy: If ``True``, transparently retry with backoff when
the server reports ``StudioBusyError``/``ConcurrencyLimitError``,
instead of raising immediately.
max_wait: Maximum total seconds to retry when *retry_on_busy* is
``True``, before re-raising the busy error.
Returns:
The execution result containing the script return value and any
console output.
Raises:
ConnectionError: If the server cannot be reached.
AuthenticationError: If the token is invalid or the IP is blocked.
ScriptSyntaxError: If the script contains a parse error.
ScriptRuntimeError: If the script raises a runtime exception.
StudioBusyError: If DAZ Studio's main thread is busy and
*retry_on_busy* is ``False`` or *max_wait* is exceeded.
ConcurrencyLimitError: If too many concurrent requests are in
flight and *retry_on_busy* is ``False`` or *max_wait* is
exceeded.
TimeoutError: If the request exceeds *timeout* seconds.
"""
payload: dict = {"script": script}
if args is not None:
payload["args"] = args
def _do():
resp = self._post("/execute", payload)
return _map_response(resp, script=script)
return self._with_busy_retry(_do, retry_on_busy, max_wait)
[docs]
def execute_file(
self, script_file: str, args: object = None, *, retry_on_busy: bool = False, max_wait: float = 30.0
) -> ExecutionResult:
"""Execute a ``.dsa`` script file that resides on the DAZ Studio host.
Args:
script_file: Absolute path to the ``.dsa`` file on the server host.
args: Optional argument passed to the script.
retry_on_busy: If ``True``, transparently retry with backoff when
the server reports ``StudioBusyError``/``ConcurrencyLimitError``,
instead of raising immediately.
max_wait: Maximum total seconds to retry when *retry_on_busy* is
``True``, before re-raising the busy error.
Returns:
The execution result.
Raises:
ConnectionError: If the server cannot be reached.
AuthenticationError: On auth failure.
ScriptSyntaxError: On parse error.
ScriptRuntimeError: On runtime error.
StudioBusyError: If DAZ Studio's main thread is busy and
*retry_on_busy* is ``False`` or *max_wait* is exceeded.
ConcurrencyLimitError: If too many concurrent requests are in
flight and *retry_on_busy* is ``False`` or *max_wait* is
exceeded.
TimeoutError: On HTTP timeout.
"""
payload: dict = {"scriptFile": script_file}
if args is not None:
payload["args"] = args
def _do():
resp = self._post("/execute", payload)
return _map_response(resp)
return self._with_busy_retry(_do, retry_on_busy, max_wait)
[docs]
def execute_async_submit(
self, script: str, args: object = None, *, report_file: str | None = None,
retry_on_busy: bool = False, max_wait: float = 30.0
) -> str:
"""Submit a script for asynchronous execution and return immediately.
Args:
script: DazScript source code.
args: Optional argument for the script.
report_file: Optional host-side JSONL file used by the script to
report structured progress, logs, and output artefacts.
retry_on_busy: If ``True``, transparently retry with backoff when
the server reports ``StudioBusyError``/``ConcurrencyLimitError``,
instead of raising immediately.
max_wait: Maximum total seconds to retry when *retry_on_busy* is
``True``, before re-raising the busy error.
Returns:
The server-assigned ``request_id`` string. Use it with
:meth:`get_request_status` or :meth:`get_request_result` to poll
for the outcome.
Raises:
ConnectionError: If the server cannot be reached.
AuthenticationError: On auth failure.
StudioBusyError: If DAZ Studio's main thread is busy and
*retry_on_busy* is ``False`` or *max_wait* is exceeded.
ConcurrencyLimitError: If too many concurrent requests are in
flight and *retry_on_busy* is ``False`` or *max_wait* is
exceeded.
"""
payload: dict = {"script": script}
if args is not None:
payload["args"] = args
if report_file is not None:
payload["reportFile"] = report_file
def _do():
resp = self._post("/execute/async", payload)
_raise_for_error(resp)
return resp.json().get("request_id", "")
return self._with_busy_retry(_do, retry_on_busy, max_wait)
[docs]
def execute_file_async_submit(
self, script_file: str, args: object = None, *, report_file: str | None = None,
retry_on_busy: bool = False, max_wait: float = 30.0
) -> str:
"""Submit a host-side ``.dsa`` file for asynchronous execution.
The file is loaded by DAZ Studio when the queued job starts, preserving
its filename for ``getScriptFileName()`` and relative ``include()``
calls. Returns the server-assigned request id immediately; use the
request status/result/cancel methods to manage its lifecycle.
"""
payload: dict = {"scriptFile": script_file}
if args is not None:
payload["args"] = args
if report_file is not None:
payload["reportFile"] = report_file
def _do():
resp = self._post("/execute/async", payload)
_raise_for_error(resp)
return resp.json().get("request_id", "")
return self._with_busy_retry(_do, retry_on_busy, max_wait)
[docs]
def execute_batch_async(
self, operations: list[dict], args: object = None, *, report_file: str | None = None
) -> str:
"""Submit multiple operations as one async request (one queue slot, one script).
Args:
operations: List of ``{"body_lines": [...], "result_expression": "..."}``
dicts — same shape as :meth:`~dazpy.Batch.add_operation`'s arguments.
args: Optional argument passed to the combined script.
report_file: Optional host-side JSONL file used for structured job
observation, as in :meth:`execute_async_submit`.
Returns:
The server-assigned ``request_id``. Poll it like any other async
request; the result's ``result`` field is a dict keyed ``"_r0"``,
``"_r1"``, ... in submission order.
"""
from ._batch import build_operations_script
pairs = [(op["body_lines"], op["result_expression"]) for op in operations]
script = build_operations_script(pairs)
return self.execute_async_submit(script, args=args, report_file=report_file)
[docs]
def get_request_status(self, request_id: str) -> dict:
"""Return the current status of an async request.
Args:
request_id: The ID returned by :meth:`execute_async_submit`.
Returns:
A dict with at least ``"status"`` and ``"observation"`` keys.
Observation contains structured progress, the bounded log tail,
and the output manifest. Possible status values:
``"queued"``, ``"running"``, ``"completed"``, ``"failed"``,
``"cancelled"``, or ``"not_found"``.
"""
resp = self._get(f"/requests/{request_id}/status")
if resp.status_code == 404:
return {"status": "not_found"}
return resp.json()
[docs]
def get_request_result(self, request_id: str, wait: bool = False, wait_timeout: int = 30) -> dict:
"""Fetch the result of a completed async request.
Args:
request_id: The ID returned by :meth:`execute_async_submit`.
wait: If ``True``, the server will long-poll until the request
completes or *wait_timeout* is reached.
wait_timeout: Maximum number of seconds the server should wait
before returning (only relevant when *wait* is ``True``).
Returns:
A dict containing ``success``, ``result``, ``output``, ``error``,
``duration_ms``, ``status``, and ``observation`` keys.
"""
params = {}
if wait:
params["wait"] = "true"
params["timeout"] = str(wait_timeout)
resp = self._get(f"/requests/{request_id}/result", params=params or None)
if resp.status_code == 404:
return {"status": "not_found"}
return resp.json()
[docs]
def list_requests(self, status: str | None = None) -> dict:
"""List all tracked async requests (script and render) with their status.
Args:
status: Optional filter, one of ``"queued"``, ``"running"``,
``"completed"``, ``"failed"``, ``"cancelled"``. Omit to list
requests in every status.
Returns:
A dict with a ``"requests"`` list (each item has ``request_id``,
``status``, ``progress``, ``submitted_at``) plus ``total`` and a
per-status count for every status value.
"""
params = {"status": status} if status else None
resp = self._get("/requests", params=params)
if resp.status_code in (401, 403):
raise AuthenticationError(f"HTTP {resp.status_code}: {resp.text[:200]}")
return resp.json()
[docs]
def cancel_request(self, request_id: str) -> bool:
"""Cancel a queued or running async request.
Args:
request_id: The ID returned by :meth:`execute_async_submit`.
Returns:
``True`` if the server confirmed cancellation, ``False`` otherwise.
"""
try:
resp = self._session.delete(
f"{self._base}/requests/{request_id}",
headers=self._headers,
timeout=self._timeout,
)
return resp.status_code == 200
except _requests.exceptions.RequestException:
return False
# ── Render ────────────────────────────────────────────────────────────────
[docs]
def render_submit(
self,
output_path: str,
*,
figure: str | None = None,
morphs: dict | None = None,
figures: list | None = None,
width: int = 0,
height: int = 0,
camera: str = "",
engine: str = "",
iray_samples: int = 0,
reset_morphs: bool = False,
retry_on_busy: bool = False,
max_wait: float = 30.0,
) -> dict:
"""Submit a render job and return immediately.
Args:
output_path: Absolute path on the DAZ Studio host to write the image.
figure: Label of the figure to configure morphs on.
morphs: Morph values to apply ``{label: value}``.
figures: List of ``{"name": ..., "morphs": {...}}`` dicts for multi-figure scenes.
width: Image width in pixels (must be paired with *height*).
height: Image height in pixels (must be paired with *width*).
camera: Camera label to render from.
engine: Render engine (``"iray"``, ``"viewport"``, ``"filament"``).
iray_samples: iRay sample count (0 = use scene default).
reset_morphs: If ``True``, reset all morphs to defaults before applying.
retry_on_busy: If ``True``, transparently retry with backoff when
the server reports ``StudioBusyError``/``ConcurrencyLimitError``,
instead of raising immediately.
max_wait: Maximum total seconds to retry when *retry_on_busy* is
``True``, before re-raising the busy error.
Returns:
A dict with ``request_id``, ``status`` (``"queued"``), and
``submitted_at`` keys.
Raises:
ConnectionError: If the server cannot be reached.
AuthenticationError: On HTTP 401/403.
StudioBusyError: If DAZ Studio's main thread is busy and
*retry_on_busy* is ``False`` or *max_wait* is exceeded.
ConcurrencyLimitError: If too many concurrent requests are in
flight and *retry_on_busy* is ``False`` or *max_wait* is
exceeded.
"""
payload: dict = {"output_path": output_path}
if width and height:
payload["width"] = width
payload["height"] = height
if camera:
payload["camera"] = camera
if engine:
payload["engine"] = engine
if iray_samples:
payload["iray_samples"] = iray_samples
if reset_morphs:
payload["reset_morphs"] = True
if figures is not None:
payload["figures"] = figures
elif figure:
payload["figure"] = figure
if morphs:
payload["morphs"] = morphs
def _do():
resp = self._post("/render", payload)
_raise_for_error(resp)
return resp.json()
return self._with_busy_retry(_do, retry_on_busy, max_wait)
[docs]
def render_batch_submit(
self, variants: list, base: dict | None = None, *, retry_on_busy: bool = False, max_wait: float = 30.0
) -> dict:
"""Submit a batch render job and return immediately.
Args:
variants: List of variant dicts, each with at least ``output_path``.
Supported keys mirror :meth:`render_submit` optional fields.
base: Optional shared defaults applied to all variants.
retry_on_busy: If ``True``, transparently retry with backoff when
the server reports ``StudioBusyError``/``ConcurrencyLimitError``,
instead of raising immediately.
max_wait: Maximum total seconds to retry when *retry_on_busy* is
``True``, before re-raising the busy error.
Returns:
A dict with ``batch_id``, ``request_ids`` (list), and ``total`` keys.
Raises:
ConnectionError: If the server cannot be reached.
AuthenticationError: On HTTP 401/403.
StudioBusyError: If DAZ Studio's main thread is busy and
*retry_on_busy* is ``False`` or *max_wait* is exceeded.
ConcurrencyLimitError: If too many concurrent requests are in
flight and *retry_on_busy* is ``False`` or *max_wait* is
exceeded.
"""
payload: dict = {"variants": variants}
if base:
payload["base"] = base
def _do():
resp = self._post("/render/batch", payload)
_raise_for_error(resp)
return resp.json()
return self._with_busy_retry(_do, retry_on_busy, max_wait)
[docs]
def render_animation_submit(
self,
output_path: str,
start_frame: int,
end_frame: int,
*,
frame_padding: int = 4,
width: int = 0,
height: int = 0,
camera: str = "",
engine: str = "",
retry_on_busy: bool = False,
max_wait: float = 30.0,
) -> dict:
"""Submit an animation render job spanning a frame range and return immediately.
Renders each frame in ``[start_frame, end_frame]`` to a separate file,
as a single trackable async request (mirrors :meth:`render_submit`'s
request-tracking shape, not :meth:`render_batch_submit`'s fan-out).
Args:
output_path: Output path pattern containing the literal token
``"{frame}"``, e.g. ``r"C:\\tmp\\anim\\frame_{frame}.png"``.
The token is replaced with the frame number, zero-padded to
*frame_padding* digits.
start_frame: First frame to render (inclusive).
end_frame: Last frame to render (inclusive).
frame_padding: Zero-padding width for the frame number (default ``4``).
width: Image width in pixels (must be paired with *height*).
height: Image height in pixels (must be paired with *width*).
camera: Camera label to render from.
engine: Render engine (``"iray"``, ``"viewport"``, ``"filament"``).
retry_on_busy: If ``True``, transparently retry with backoff when
the server reports ``StudioBusyError``/``ConcurrencyLimitError``,
instead of raising immediately.
max_wait: Maximum total seconds to retry when *retry_on_busy* is
``True``, before re-raising the busy error.
Returns:
A dict with ``request_id``, ``status`` (``"queued"``), and
``submitted_at`` keys.
Raises:
ConnectionError: If the server cannot be reached.
AuthenticationError: On HTTP 401/403.
StudioBusyError: If DAZ Studio's main thread is busy and
*retry_on_busy* is ``False`` or *max_wait* is exceeded.
ConcurrencyLimitError: If too many concurrent requests are in
flight and *retry_on_busy* is ``False`` or *max_wait* is
exceeded.
"""
payload: dict = {
"output_path": output_path,
"start_frame": start_frame,
"end_frame": end_frame,
"frame_padding": frame_padding,
}
if width and height:
payload["width"] = width
payload["height"] = height
if camera:
payload["camera"] = camera
if engine:
payload["engine"] = engine
def _do():
resp = self._post("/render/animation", payload)
_raise_for_error(resp)
return resp.json()
return self._with_busy_retry(_do, retry_on_busy, max_wait)
[docs]
def cancel_render(self, request_id: str) -> bool:
"""Cancel a queued or running render job.
Args:
request_id: The ``request_id`` from :meth:`render_submit` or a
:class:`~dazpy._render_api.RenderResult` with ``wait=False``.
Returns:
``True`` if the server confirmed cancellation, ``False`` otherwise
(already finished, not found, or connection error).
"""
try:
resp = self._session.post(
f"{self._base}/render/{request_id}/cancel",
headers=self._headers,
timeout=self._timeout,
)
return resp.status_code == 200
except _requests.exceptions.RequestException:
return False
[docs]
def stream_render_progress(self, request_id: str, stream_timeout: float = 305.0) -> "object | None":
"""Open the SSE progress stream for a render request.
Returns a streaming :class:`requests.Response` on success, or ``None``
if the endpoint is unavailable. Callers must close the response when done.
"""
try:
resp = self._session.get(
f"{self._base}/render/{request_id}/progress",
headers=self._headers,
stream=True,
timeout=stream_timeout,
)
return resp if resp.status_code == 200 else None
except _requests.exceptions.RequestException:
return None
[docs]
def stream_scene_events(
self,
categories: "list[str] | None" = None,
stream_timeout: "float | None" = None,
) -> "object | None":
"""Open the SSE stream for general scene-change events (GET /scene/events).
Args:
categories: Optional subset of event categories to subscribe to
(``"node"``, ``"skeleton"``, ``"light"``, ``"camera"``,
``"selection"``, ``"scene"``, ``"time"``, ``"render"``).
``None`` (default) subscribes to all categories.
stream_timeout: Socket timeout in seconds. ``None`` (default)
waits indefinitely — the server sends a keepalive comment
every 15 seconds, so the connection never idles out.
Returns:
A streaming :class:`requests.Response` on success, or ``None``
if the endpoint is unavailable. Callers must close the response
when done (e.g. via a ``with`` statement).
"""
try:
params = {"filter": ",".join(categories)} if categories else None
resp = self._session.get(
f"{self._base}/scene/events",
headers=self._headers,
params=params,
stream=True,
timeout=stream_timeout,
)
return resp if resp.status_code == 200 else None
except _requests.exceptions.RequestException:
return None
# ── USD export ────────────────────────────────────────────────────────────
[docs]
def export_usd_submit(
self,
output_path: str,
*,
include_geometry: bool = True,
include_materials: bool = True,
include_skeleton: bool = False,
include_morphs: bool = False,
include_lights: bool = False,
include_camera: bool = False,
) -> dict:
"""Submit a USD export job and return immediately.
Args:
output_path: Absolute path on the DAZ Studio host where the
``.usda`` file should be written.
include_geometry: Export mesh geometry (default ``True``).
include_materials: Export PBR material prims (default ``True``).
include_skeleton: Export UsdSkel armature and skin weights.
include_morphs: Export active morphs as UsdSkelBlendShape prims.
include_lights: Export scene lights using UsdLux prims.
include_camera: Export the active render camera as UsdGeomCamera.
Returns:
A dict with ``job_id``, ``status`` (``"queued"``), and
``submittedAt`` keys.
Raises:
ConnectionError: If the server cannot be reached.
AuthenticationError: On HTTP 401/403.
"""
payload = {
"outputPath": output_path,
"includeGeometry": include_geometry,
"includeMaterials": include_materials,
"includeSkeleton": include_skeleton,
"includeMorphs": include_morphs,
"includeLights": include_lights,
"includeCamera": include_camera,
}
resp = self._post("/export/usd", payload)
if resp.status_code in (401, 403):
raise AuthenticationError(f"HTTP {resp.status_code}: {resp.text[:200]}")
return resp.json()
[docs]
def get_usd_export_status(self, job_id: str) -> dict:
"""Poll the status of a USD export job.
Args:
job_id: The ID returned by :meth:`export_usd_submit`.
Returns:
A dict with ``job_id``, ``status``, and (on completion)
``outputPath`` or ``error`` keys.
Raises:
ConnectionError: If the server cannot be reached.
AuthenticationError: On HTTP 401/403.
"""
resp = self._get(f"/export/usd/{job_id}")
if resp.status_code in (401, 403):
raise AuthenticationError(f"HTTP {resp.status_code}: {resp.text[:200]}")
if resp.status_code == 404:
return {"job_id": job_id, "status": "not_found"}
return resp.json()
# ── Server health ─────────────────────────────────────────────────────────
[docs]
def status(self) -> dict:
"""Return the server status dict from ``GET /status``.
Raises:
AuthenticationError: On HTTP 401/403.
"""
resp = self._get("/status")
if resp.status_code in (401, 403):
raise AuthenticationError(f"HTTP {resp.status_code}: {resp.text[:200]}")
return resp.json()
[docs]
def health(self) -> dict:
"""Return the health check dict from ``GET /health``.
Raises:
AuthenticationError: On HTTP 401/403.
"""
resp = self._get("/health")
if resp.status_code in (401, 403):
raise AuthenticationError(f"HTTP {resp.status_code}: {resp.text[:200]}")
return resp.json()
[docs]
def metrics(self) -> dict:
"""Return the metrics dict from ``GET /metrics``.
Raises:
AuthenticationError: On HTTP 401/403.
"""
resp = self._get("/metrics")
if resp.status_code in (401, 403):
raise AuthenticationError(f"HTTP {resp.status_code}: {resp.text[:200]}")
return resp.json()