make_async_snapshot_request() — anthropic-sdk-python Function Reference

Architecture documentation for the make_async_snapshot_request() function in snapshots.py from the anthropic-sdk-python codebase.

Function python AnthropicClient AsyncAPI

Entity Profile

AnthropicClient→ AsyncAPI→ make_async_snapshot_request() — anthropic-sdk-python Function Reference

Dependency Diagram

graph TD
  f6132c7b_5c13_10c7_101a_9e6f86b739af["make_async_snapshot_request()"]
  e7250a77_f5e0_e4de_3e1b_f2b040399c23["snapshots.py"]
  f6132c7b_5c13_10c7_101a_9e6f86b739af -->|defined in| e7250a77_f5e0_e4de_3e1b_f2b040399c23
  style f6132c7b_5c13_10c7_101a_9e6f86b739af fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

tests/lib/snapshots.py lines 199–259

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

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

        respx_mock.stop()

        client = AsyncAnthropic(
            http_client=httpx.AsyncClient(
                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"},
            )

        respx_mock.post(path).mock(side_effect=get_response)

        client = mock_client

    result = await func(client)

    if not live:
        return result

    await client.close()

    if len(collected) == 1:
        assert collected[0] == content_snapshot
    else:
        assert collected == content_snapshot

    return result