snapshots.py — anthropic-sdk-python Source File

Source Code

from __future__ import annotations

import os
import json
from typing import Any, List, Callable, Awaitable, cast
from typing_extensions import TypeVar

import httpx
from respx import MockRouter
from inline_snapshot import outsource, get_snapshot_value

from anthropic import Anthropic, AsyncAnthropic
from anthropic._utils._utils import is_list

_T = TypeVar("_T")


def make_snapshot_request(
    func: Callable[[Anthropic], _T],
    *,
    content_snapshot: Any,
    respx_mock: MockRouter,
    mock_client: Anthropic,
    path: str,
) -> _T:
    live = os.environ.get("ANTHROPIC_LIVE") == "1"
    collected: list[str] = []
    if live:

        def _on_response(response: httpx.Response) -> None:
            collected.append(json.dumps(json.loads(response.read())))

        respx_mock.stop()

        client = Anthropic(
            http_client=httpx.Client(
                event_hooks={
                    "response": [_on_response],
                }
            )
        )
    else:
        responses = get_snapshot_value(content_snapshot)

        if not is_list(responses):
            responses = [responses]

        curr = 0

        def get_response(_request: httpx.Request) -> httpx.Response:
            nonlocal curr
            content = responses[curr]
            assert isinstance(content, str)

            curr += 1
            return httpx.Response(
                200,
                content=content,
                headers={"content-type": "application/json"},
            )
// ... (200 more lines)