truegrain

353@dataclass(frozen=True)
354class AuditEvent:
355    """One decision the engine recorded.
356
357    ``refused`` and ``denied`` are deliberately separate. Denied is access:
358    this caller may not read something. Refused is correctness: nobody can be
359    told this accurately, and ``hint`` names a question that can be. Counting
360    them together would report every fan-out as an access incident.
361
362    Carries no filter values, no compiled SQL and no result rows. ``sql_hash``
363    identifies the statement without disclosing it.
364    """
365
366    #: RFC 3339.
367    time: str
368    #: One of ``allowed``, ``refused``, ``denied``, ``compiled``, ``error``.
369    decision: str
370    #: Who asked. ``anonymous`` on an engine without authentication.
371    identity: str = "anonymous"
372    #: Which refusal, when ``decision`` is ``refused``.
373    code: str = ""
374    #: What the caller should do about a refusal: ``modify``, ``later`` or
375    #: ``never``. Empty for anything that is not a refusal.
376    retry: str = ""
377    reason: str = ""
378    #: What would answer instead. The actionable half of a refusal.
379    hint: str = ""
380    #: What was asked for. Names only, never values.
381    metrics: tuple[str, ...] = ()
382    dimensions: tuple[str, ...] = ()
383    namespace: str = ""
384    model_name: str = ""
385    #: The workspace digest that answered, tying a number to a model revision.
386    model_version: str = ""
387    dialect: str = ""
388    #: Semantic fields refused on a denial. Never returned to the denied caller.
389    denied_fields: tuple[str, ...] = ()
390    sql_hash: str = ""
391    #: The warehouse's own identifier, so two logs can be joined.
392    job_id: str = ""
393    row_count: int = 0
394    #: What the warehouse says the query cost. Zero means not reported rather
395    #: than free: DuckDB bills nobody and reports nothing, and the two are
396    #: indistinguishable here on purpose.
397    bytes_billed: int = 0
398    duration_ms: int = 0
399    #: The failure, when ``decision`` is ``error``.
400    error: str = ""
401    raw: dict[str, Any] = field(default_factory=dict, repr=False)
402
403    @classmethod
404    def parse(cls, payload: dict[str, Any]) -> "AuditEvent":
405        retry = _get(payload, "retry", "")
406        return cls(
407            time=_get(payload, "time", ""),
408            # An unrecognised decision reads as ``error`` rather than as an
409            # empty string, so a caller branching on it cannot silently treat
410            # something the engine meant as significant as nothing at all.
411            decision=_get(payload, "decision", "") or "error",
412            identity=_get(payload, "identity", "") or "anonymous",
413            code=_get(payload, "refusal_code", ""),
414            retry=retry if retry in ("modify", "later", "never") else "",
415            reason=_get(payload, "reason", ""),
416            hint=_get(payload, "hint", ""),
417            metrics=tuple(_get(payload, "metrics", [])),
418            dimensions=tuple(_get(payload, "dimensions", [])),
419            namespace=_get(payload, "namespace", ""),
420            model_name=_get(payload, "model_name", ""),
421            model_version=_get(payload, "model_version", ""),
422            dialect=_get(payload, "dialect", ""),
423            denied_fields=tuple(_get(payload, "denied_fields", [])),
424            sql_hash=_get(payload, "sql_hash", ""),
425            job_id=_get(payload, "job_id", ""),
426            row_count=int(_get(payload, "row_count", 0)),
427            bytes_billed=int(_get(payload, "bytes_billed", 0)),
428            duration_ms=int(_get(payload, "duration_ms", 0)),
429            error=_get(payload, "error", ""),
430            raw=payload,
431        )
432
433    def is_refusal(self) -> bool:
434        """True when the engine declined for correctness rather than access."""
435        return self.decision == "refused"
436
437    def is_denial(self) -> bool:
438        """True when the engine declined because this caller may not read something."""
439        return self.decision == "denied"

442@dataclass(frozen=True)
443class AuditPage:
444    """A window of recorded decisions, newest first.
445
446    A window rather than the whole record: the engine holds a bounded buffer in
447    memory. An engine started with an audit file keeps everything there, and
448    ``note`` says so.
449    """
450
451    events: tuple[AuditEvent, ...] = ()
452    #: How many are in this page, as the engine counted them.
453    count: int = 0
454    note: str = ""
455    raw: dict[str, Any] = field(default_factory=dict, repr=False)
456
457    @classmethod
458    def parse(cls, payload: dict[str, Any]) -> "AuditPage":
459        events = tuple(AuditEvent.parse(e) for e in _get(payload, "events", []))
460        return cls(
461            events=events,
462            # The engine's own count, falling back to what arrived. They agree,
463            # and preferring the engine's means a disagreement is visible rather
464            # than papered over by recounting.
465            count=int(_get(payload, "count", len(events))),
466            note=_get(payload, "note", ""),
467            raw=payload,
468        )
469
470    def __iter__(self) -> Iterator[AuditEvent]:
471        return iter(self.events)
472
473    def __len__(self) -> int:
474        return len(self.events)
475
476    def refusals(self) -> tuple[AuditEvent, ...]:
477        """Only the refusals, which is what an operator opens this to read."""
478        return tuple(e for e in self.events if e.is_refusal())

758@dataclass(frozen=True)
759class Change:
760    """What one request compiled to before and after a reload."""
761
762    before: str = ""
763    after: str = ""
764
765    @classmethod
766    def parse(cls, payload: dict[str, Any]) -> "Change":
767        return cls(before=_get(payload, "before", ""), after=_get(payload, "after", ""))

 72class Client:
 73    """A connection to one semantic engine.
 74
 75    Args:
 76        base_url: Where the engine is served, for example
 77            ``http://127.0.0.1:8080``.
 78        token: Bearer token identifying the workload. Read it from the
 79            environment; never hard-code one. Omit it only against a local
 80            engine running without authentication.
 81        timeout: Seconds to wait for a response.
 82        user_agent: Overrides the identifying header, which is useful when you
 83            want the audit log to distinguish one agent from another.
 84
 85    Example:
 86        >>> from truegrain import Client, filters
 87        >>> c = Client("http://127.0.0.1:8080", token=os.environ["SEMANTIC_TOKEN"])
 88        >>> result = c.query(
 89        ...     metrics=["sales.order_revenue"],
 90        ...     dimensions=["sales.customers.region"],
 91        ...     filters=[filters.eq("sales.orders.status", "shipped")],
 92        ... )
 93        >>> result.to_dataframe()
 94    """
 95
 96    def __init__(
 97        self,
 98        base_url: str,
 99        token: str | None = None,
100        timeout: float = DEFAULT_TIMEOUT,
101        user_agent: str = "truegrain-python/1.0",
102    ) -> None:
103        if not base_url:
104            raise ValueError("base_url is required")
105        self.base_url = base_url.rstrip("/")
106        self._token = token
107        self.timeout = timeout
108        self.user_agent = user_agent
109
110    @classmethod
111    def from_env(
112        cls,
113        url_var: str = "SEMANTIC_URL",
114        token_var: str = "SEMANTIC_TOKEN",
115        **kwargs: Any,
116    ) -> "Client":
117        """Build a client from environment variables.
118
119        Keeps the token out of source and out of the process list, which is
120        where a token passed as a command-line flag ends up.
121        """
122        url = os.environ.get(url_var)
123        if not url:
124            raise SemanticError(f"{url_var} is not set")
125        return cls(url, token=os.environ.get(token_var), **kwargs)
126
127    # ---------- metadata ----------
128
129    def health(self) -> Health:
130        """What this deployment enforces, including what it does not.
131
132        Worth reading before trusting the layer with anything sensitive:
133        ``health().enforcement_notes`` states the gaps in plain language.
134        """
135        return Health.parse(self._get("/v1/health"))
136
137    def model_version(self) -> str:
138        """The workspace digest currently served.
139
140        Every query result carries this too. Two results with the same digest
141        were produced by exactly the same definitions.
142        """
143        return str(self._get("/v1/model/version").get("model_version", ""))
144
145    def namespaces(self) -> list[Namespace]:
146        """The namespaces in the workspace, their owners and availability."""
147        payload = self._get("/v1/namespaces")
148        return [Namespace.parse(n) for n in payload.get("namespaces", [])]
149
150    def metrics(self, search: str | None = None) -> list[Metric]:
151        """Every metric this identity may read.
152
153        Args:
154            search: Case-insensitive substring over name and description.
155        """
156        params = {"search": search} if search else None
157        payload = self._get("/v1/metrics", params)
158        return [Metric.parse(m) for m in payload.get("metrics", [])]
159
160    def metric(self, name: str) -> Metric:
161        """One metric's definition and the dimensions legal for it.
162
163        Args:
164            name: Qualified as ``namespace.metric``, or bare when unambiguous
165                across the workspace.
166        """
167        if not name:
168            raise ValueError("metric name is required")
169        return Metric.parse(self._get(f"/v1/metrics/{urllib.parse.quote(name, safe='')}"))
170
171    def dimensions(self, metric: str | None = None) -> list[Dimension]:
172        """Dimensions available for grouping and filtering.
173
174        Args:
175            metric: Restrict to dimensions valid for this metric, which is
176                almost always what you want.
177        """
178        params = {"metric": metric} if metric else None
179        payload = self._get("/v1/dimensions", params)
180        return [Dimension.parse(d) for d in payload.get("dimensions", [])]
181
182    # ---------- query ----------
183
184    def query(
185        self,
186        metrics: Iterable[str],
187        dimensions: Iterable[str] | None = None,
188        filters: Iterable[Mapping[str, Any]] | None = None,
189        grain: str | None = None,
190        limit: int | None = None,
191        order_by: Iterable[Mapping[str, Any]] | None = None,
192    ) -> Result:
193        """Answer a question and return rows.
194
195        Args:
196            metrics: Metric names. Metrics at different grains, or in different
197                namespaces, are aggregated separately and joined on the shared
198                dimensions.
199            dimensions: ``dataset.field`` or ``namespace.dataset.field``.
200            filters: Structured filters. Use :mod:`truegrain.filters` to
201                build them.
202            grain: Time bucket for the selected time dimension.
203            limit: Maximum rows. Defaults to the server's limit.
204            order_by: ``[{"field": "order_revenue", "desc": True}]``.
205
206        Raises:
207            Refused: The engine declined. Check ``err.retry`` before retrying.
208        """
209        return Result.parse(
210            self._post("/v1/query", self._request_body(metrics, dimensions, filters, grain, limit, order_by))
211        )
212
213    def compile(
214        self,
215        metrics: Iterable[str],
216        dimensions: Iterable[str] | None = None,
217        filters: Iterable[Mapping[str, Any]] | None = None,
218        grain: str | None = None,
219        limit: int | None = None,
220        order_by: Iterable[Mapping[str, Any]] | None = None,
221    ) -> Compiled:
222        """Return the SQL a request compiles to, without running it.
223
224        A dry run is still governed: a request you may not run returns a refusal
225        and no SQL, and the inspection is audited. It is a way to see what a
226        query would do, not a way around the gate.
227        """
228        return Compiled.parse(
229            self._post("/v1/compile", self._request_body(metrics, dimensions, filters, grain, limit, order_by))
230        )
231
232    # ---------- asynchronous execution ----------
233
234    def run(
235        self,
236        metrics: Iterable[str],
237        dimensions: Iterable[str] | None = None,
238        filters: Iterable[Mapping[str, Any]] | None = None,
239        grain: str | None = None,
240        limit: int | None = None,
241        order_by: Iterable[Mapping[str, Any]] | None = None,
242        max_wait: float = 900.0,
243        page_size: int | None = None,
244    ) -> Result:
245        """Run a query that may take longer than an HTTP request should.
246
247        Submits the query, polls until it finishes, collects every page, and
248        returns one :class:`Result`. Use this instead of :meth:`query` whenever
249        the warehouse might be slow: a large BigQuery scan routinely outlives
250        the default timeout of an agent framework, and a synchronous call that
251        times out leaves the query running and billable with nobody reading it.
252
253        The governance gate runs during submission, so a request this identity
254        may not make raises :class:`Refused` immediately rather than after a
255        wait.
256
257        Args:
258            max_wait: Seconds to wait before giving up. On expiry the job is
259                cancelled rather than left running, so an abandoned wait does
260                not keep spending warehouse time.
261            page_size: Rows per page while collecting. The default suits most
262                results; lower it when rows are wide.
263
264        Raises:
265            Refused: The engine declined, or the query failed. Check
266                ``err.retry`` before retrying.
267            SemanticError: The wait expired, or the job was cancelled.
268        """
269        job = self.submit(metrics, dimensions, filters, grain, limit, order_by)
270        job = self.wait(job.job_id, max_wait=max_wait, page_size=page_size)
271        job.raise_for_state()
272
273        # Walk the remaining pages. row_count is the size of the whole result,
274        # so it bounds the walk without the loop having to trust the cursor to
275        # eventually come back empty.
276        rows = list(job.rows)
277        cursor = job.next_cursor
278        total = job.row_count or None
279        while cursor and (total is None or len(rows) < total):
280            page = self.job(job.job_id, cursor=cursor, page_size=page_size)
281            if not page.rows:
282                break
283            rows.extend(page.rows)
284            cursor = page.next_cursor
285        return job.as_result(tuple(rows))
286
287    def submit(
288        self,
289        metrics: Iterable[str],
290        dimensions: Iterable[str] | None = None,
291        filters: Iterable[Mapping[str, Any]] | None = None,
292        grain: str | None = None,
293        limit: int | None = None,
294        order_by: Iterable[Mapping[str, Any]] | None = None,
295    ) -> Job:
296        """Start a query in the background and return immediately.
297
298        The returned job already carries ``compiled_sql``, because compilation
299        happened synchronously. A job coming back at all means the query was
300        authorized, not merely accepted.
301
302        Prefer :meth:`run` unless you genuinely need to do something else while
303        the query runs.
304        """
305        return Job.parse(
306            self._post(
307                "/v1/jobs",
308                self._request_body(metrics, dimensions, filters, grain, limit, order_by),
309            )
310        )
311
312    def job(self, job_id: str, cursor: str | None = None, page_size: int | None = None) -> Job:
313        """Poll a job and read one page of its rows.
314
315        Args:
316            cursor: From a previous page's ``next_cursor``. Omit for the first.
317            page_size: Rows in this page.
318        """
319        if not job_id:
320            raise ValueError("job_id is required")
321        params: dict[str, str] = {}
322        if cursor:
323            params["cursor"] = cursor
324        if page_size is not None:
325            params["page_size"] = str(page_size)
326        return Job.parse(
327            self._get(f"/v1/jobs/{urllib.parse.quote(job_id, safe='')}", params or None)
328        )
329
330    def cancel_job(self, job_id: str) -> Job:
331        """Stop a running query. Idempotent."""
332        if not job_id:
333            raise ValueError("job_id is required")
334        return Job.parse(
335            self._delete(f"/v1/jobs/{urllib.parse.quote(job_id, safe='')}")
336        )
337
338    def wait(
339        self,
340        job_id: str,
341        max_wait: float = 900.0,
342        page_size: int | None = None,
343    ) -> Job:
344        """Poll until a job reaches a terminal state.
345
346        Backs off from a fast first poll to a two second ceiling, so a query
347        that finishes quickly is not delayed and a slow one is not hammered.
348
349        On expiry the job is cancelled before raising, so giving up here also
350        stops the warehouse work.
351        """
352        deadline = time.monotonic() + max_wait
353        interval = 0.25
354        while True:
355            job = self.job(job_id, page_size=page_size)
356            if job.is_done:
357                return job
358            if time.monotonic() >= deadline:
359                # Leaving it running would keep spending warehouse time for an
360                # answer this caller has already stopped waiting for.
361                try:
362                    self.cancel_job(job_id)
363                except SemanticError:
364                    pass
365                raise SemanticError(
366                    f"job {job_id} did not finish within {max_wait:g}s and was cancelled"
367                )
368            time.sleep(min(interval, max(0.0, deadline - time.monotonic())))
369            interval = min(interval * 1.5, 2.0)
370
371    # ---------- audit ----------
372
373    def audit(self, limit: int | None = None, decision: str | None = None) -> AuditPage:
374        """Read the decisions the engine recently made, newest first.
375
376        Not served unless an operator started the engine with ``-audit-readers``
377        naming who may read it, because the record carries every caller's
378        activity. An engine with nobody named answers 404 and a caller who is
379        authenticated but not named gets 403, both as :class:`Refused`. Handle
380        both: this is one of the few calls that can fail for reasons that have
381        nothing to do with the request.
382
383        It is a bounded window rather than the whole record. An engine writing
384        an audit file keeps everything there.
385
386        Args:
387            limit: Most recent decisions to return, up to 200.
388            decision: Return only this decision. ``refused`` is the useful one.
389        """
390        params: dict[str, str] = {}
391        if limit is not None:
392            params["limit"] = str(limit)
393        if decision:
394            params["decision"] = decision
395        return AuditPage.parse(self._get("/v1/audit", params))
396
397    # ---------- what the engine says about itself ----------
398    #
399    # Everything here is metadata. None of it reads a row, which is why all but
400    # reload need only read:model, and why a pipeline that checks a model does
401    # not also need a credential that can read the warehouse.
402    #
403    # Four of them answer 404 rather than an empty success, and the difference
404    # matters enough to say once: an engine with no test suite is not an engine
405    # whose tests pass, an engine that has never reloaded is not an engine in
406    # which nothing changed, and an engine checking nothing on a timer has no
407    # history rather than a clean one. Each arrives as :class:`Refused`.
408
409    def doctor(self) -> Diagnosis:
410        """Ask the warehouse whether the model is still true.
411
412        Everything else validates the model against itself: the YAML parses,
413        the joins resolve, the expressions compile. None of that notices that
414        somebody dropped a column last Tuesday, and the first sign of that is
415        usually a caller getting an error, which is the most expensive place to
416        find it.
417
418        Reports rather than refuses, so an exception here means the check could
419        not run. Read :attr:`Diagnosis.ok` for the verdict and
420        :attr:`Diagnosis.skipped` for the case where nothing could be checked.
421        """
422        return Diagnosis.parse(self._get("/v1/doctor"))
423
424    def doctor_history(self) -> DoctorHistory:
425        """What the scheduled warehouse check has seen, oldest first.
426
427        Drift is found by looking regularly, not by looking once, which is how
428        "when did this start" stays answerable. An engine started without
429        ``-doctor-every`` has nothing scheduled and answers 404 as
430        :class:`Refused`.
431        """
432        return DoctorHistory.parse(self._get("/v1/doctor/history"))
433
434    def run_tests(self) -> TestReport:
435        """Assert what this model answers.
436
437        ``validate`` says the model holds together and :meth:`diff` says a
438        number changed. Neither says a number was ever right.
439
440        Check :attr:`TestReport.withheld` as well as ``ok``. A credential
441        without ``run:query`` cannot cause warehouse execution, so cases that
442        would are withheld and counted rather than run or silently dropped, and
443        a caller reading only ``ok`` would conclude a suite passed when half of
444        it never ran.
445        """
446        return TestReport.parse(self._post("/v1/tests", {}))
447
448    def policy(self) -> Policy:
449        """What this engine enforces, and what it does not.
450
451        Says nothing about who is allowed what. For that, and only about
452        yourself, use :meth:`explain_policy`.
453        """
454        return Policy.parse(self._get("/v1/policy"))
455
456    def explain_policy(self, metric: str) -> PolicyExplanation:
457        """What you may read of a metric, and why.
458
459        Answers "why can I not group by that column" without running a query
460        and being denied. For the calling identity only, which is a security
461        property rather than a limitation: an endpoint that reported another
462        identity's access would publish the policy it was configured to
463        enforce.
464
465        Args:
466            metric: The qualified name, as :meth:`metrics` reports it.
467        """
468        if not metric:
469            # Refused here rather than sent, so a caller who forgot the
470            # argument reads that instead of a 400 naming a field they did not
471            # write.
472            raise ValueError("name the metric to explain")
473        return PolicyExplanation.parse(self._post("/v1/policy/explain", {"metric": metric}))
474
475    def diff(self) -> Diff:
476        """Whether the last model reload moved a number.
477
478        This compares the model being served against the one served before it,
479        which is the comparison nobody can make from outside the process. An
480        engine that has served only one model answers 404 as :class:`Refused`,
481        because that is a different answer from nothing having changed and only
482        one of them is reassuring.
483        """
484        return Diff.parse(self._get("/v1/diff"))
485
486    def reload(self) -> str:
487        """Tell the engine to re-read its model source.
488
489        Returns ``reading`` or ``already running``. Both mean the caller got
490        what they asked for; treating the second as a failure would retry a
491        sync that is already under way.
492
493        It carries no model, deliberately: this means "look now", not "install
494        this". The engine already follows git, so the only thing this changes
495        is the wait. A method that accepted a model would be a second way into
496        production, one that skips the pull request, the checks and the diff
497        that reports which numbers move.
498
499        Needs the ``deploy:model`` scope, which ``read:model`` and
500        ``run:query`` never imply.
501
502        Returning does not mean the new model is serving. A sync is not
503        instant, and reporting a commit before the swap happened would be a
504        claim a pipeline then asserts as fact. Poll :meth:`health` and read
505        ``origin.commit`` to know when the new model is the one answering.
506
507        An engine reading from a path has nothing to re-read and answers 404. A
508        model that fails to load is not a failure of this call either: the
509        engine keeps serving the previous one and says so in a 502. Both are
510        :class:`Refused`.
511        """
512        payload = self._post("/v1/reload", {})
513        return str(payload.get("status", ""))
514
515    # ---------- internals ----------
516
517    @staticmethod
518    def _request_body(
519        metrics: Iterable[str],
520        dimensions: Iterable[str] | None,
521        filters: Iterable[Mapping[str, Any]] | None,
522        grain: str | None,
523        limit: int | None,
524        order_by: Iterable[Mapping[str, Any]] | None,
525    ) -> dict[str, Any]:
526        names = list(metrics)
527        if not names:
528            raise ValueError("at least one metric is required")
529        body: dict[str, Any] = {"metrics": names}
530        # Omitted rather than sent empty: the server rejects unknown and
531        # malformed fields, and an empty list is not the same as absent.
532        if dimensions:
533            body["dimensions"] = list(dimensions)
534        if filters:
535            body["filters"] = [dict(f) for f in filters]
536        if grain:
537            body["grain"] = grain
538        if limit is not None:
539            body["limit"] = limit
540        if order_by:
541            body["order_by"] = [dict(o) for o in order_by]
542        return body
543
544    def _headers(self) -> dict[str, str]:
545        headers = {"Accept": "application/json", "User-Agent": self.user_agent}
546        if self._token:
547            headers["Authorization"] = f"Bearer {self._token}"
548        return headers
549
550    def _get(self, path: str, params: Mapping[str, str] | None = None) -> dict[str, Any]:
551        url = self.base_url + path
552        if params:
553            url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v})
554        return self._send(urllib.request.Request(url, headers=self._headers(), method="GET"))
555
556    def _post(self, path: str, body: Mapping[str, Any]) -> dict[str, Any]:
557        headers = self._headers()
558        headers["Content-Type"] = "application/json"
559        payload = json.dumps(body).encode("utf-8")
560        request = urllib.request.Request(
561            self.base_url + path, data=payload, headers=headers, method="POST"
562        )
563        return self._send(request)
564
565    def _delete(self, path: str) -> dict[str, Any]:
566        return self._send(
567            urllib.request.Request(self.base_url + path, headers=self._headers(), method="DELETE")
568        )
569
570    def _send(self, request: urllib.request.Request) -> dict[str, Any]:
571        try:
572            with urllib.request.urlopen(request, timeout=self.timeout) as response:
573                return self._decode(response.read(), response.status)
574        except urllib.error.HTTPError as exc:
575            # A refusal arrives as an error status with a JSON body. It is an
576            # answer, not a transport failure, so it is raised as Refused with
577            # the code and retry class intact.
578            raw = exc.read()
579            if exc.code == 401:
580                raise Unauthorized(self._message(raw, "credentials were not accepted")) from None
581            try:
582                payload = json.loads(raw.decode("utf-8"))
583            except (ValueError, UnicodeDecodeError):
584                raise TransportError(
585                    f"HTTP {exc.code} from {request.full_url}: {self._message(raw, exc.reason)}"
586                ) from None
587            if isinstance(payload, dict) and "code" in payload:
588                raise Refused.from_payload(payload, exc.code) from None
589            raise TransportError(f"HTTP {exc.code} from {request.full_url}: {payload}") from None
590        except urllib.error.URLError as exc:
591            raise TransportError(f"cannot reach {request.full_url}: {exc.reason}") from None
592        except TimeoutError:
593            raise TransportError(
594                f"{request.full_url} did not respond within {self.timeout:g}s"
595            ) from None
596
597    @staticmethod
598    def _decode(raw: bytes, status: int) -> dict[str, Any]:
599        try:
600            payload = json.loads(raw.decode("utf-8"))
601        except (ValueError, UnicodeDecodeError) as exc:
602            raise TransportError(f"HTTP {status} body was not JSON: {exc}") from None
603        if not isinstance(payload, dict):
604            raise TransportError(f"HTTP {status} body was not a JSON object")
605        return payload
606
607    @staticmethod
608    def _message(raw: bytes, fallback: Any) -> str:
609        text = raw.decode("utf-8", errors="replace").strip()
610        return text or str(fallback)
611
612    def __repr__(self) -> str:
613        return f"Client(base_url={self.base_url!r}, authenticated={bool(self._token)})"