(self)
| 319 | } |
| 320 | |
| 321 | def _usage_snapshot(self) -> dict[str, dict[str, int]]: |
| 322 | return { |
| 323 | "last_request": { |
| 324 | "message_count": self._last_request_metrics["message_count"], |
| 325 | "text_part_count": self._last_request_metrics["text_part_count"], |
| 326 | "image_part_count": self._last_request_metrics["image_part_count"], |
| 327 | "input_tokens": self._last_usage_metrics["input_tokens"], |
| 328 | "cached_input_tokens": self._last_usage_metrics["cached_input_tokens"], |
| 329 | }, |
| 330 | "last_response": dict(self._last_usage_metrics), |
| 331 | "cumulative_request": { |
| 332 | "message_count": self._cumulative_request_metrics["message_count"], |
| 333 | "text_part_count": self._cumulative_request_metrics["text_part_count"], |
| 334 | "image_part_count": self._cumulative_request_metrics["image_part_count"], |
| 335 | "input_tokens": self._cumulative_usage_metrics["input_tokens"], |
| 336 | "cached_input_tokens": self._cumulative_usage_metrics["cached_input_tokens"], |
| 337 | }, |
| 338 | "cumulative_response": dict(self._cumulative_usage_metrics), |
| 339 | } |
| 340 | |
| 341 | def _log_gateway_error(self, *, event: str, attempt: int, error: BaseException) -> None: |
| 342 | response = getattr(error, "response", None) |
no outgoing calls
no test coverage detected