TestAsyncBatches Class — anthropic-sdk-python Architecture

Architecture documentation for the TestAsyncBatches class in test_batches.py from the anthropic-sdk-python codebase.

Class python AnthropicClient

Entity Profile

AnthropicClient→ TestAsyncBatches Class — anthropic-sdk-python Architecture

Dependency Diagram

graph TD
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5["TestAsyncBatches"]
  3e107774_1462_d161_d688_0579282f7e10["MessageBatch"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|extends| 3e107774_1462_d161_d688_0579282f7e10
  6f0c2763_2404_28c1_053f_f98121dd7d2a["DeletedMessageBatch"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|extends| 6f0c2763_2404_28c1_053f_f98121dd7d2a
  dd332f9f_baa1_66e7_daf2_701c347cd3bf["test_batches.py"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|defined in| dd332f9f_baa1_66e7_daf2_701c347cd3bf
  100cbb66_6176_9023_7f49_59ae711618ed["test_method_create()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| 100cbb66_6176_9023_7f49_59ae711618ed
  56868e9d_3c40_e304_5c76_242d671fdad1["test_raw_response_create()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| 56868e9d_3c40_e304_5c76_242d671fdad1
  b99936ef_2baa_e42a_ece2_2f01622a7ed3["test_streaming_response_create()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| b99936ef_2baa_e42a_ece2_2f01622a7ed3
  bd7f2396_050f_20e5_27e9_36bb87753824["test_method_retrieve()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| bd7f2396_050f_20e5_27e9_36bb87753824
  4aa303cf_c2f7_f59b_5a33_9b24b11b9c14["test_raw_response_retrieve()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| 4aa303cf_c2f7_f59b_5a33_9b24b11b9c14
  9040410f_b8ac_2329_7fd1_207396394a65["test_streaming_response_retrieve()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| 9040410f_b8ac_2329_7fd1_207396394a65
  9dd21974_21f4_fcd4_daa0_400b4598e969["test_path_params_retrieve()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| 9dd21974_21f4_fcd4_daa0_400b4598e969
  1821c5eb_8f80_e6bd_1e0c_65dd7714fba8["test_method_list()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| 1821c5eb_8f80_e6bd_1e0c_65dd7714fba8
  44606dd3_cb17_038d_86d1_0dd536e91d42["test_method_list_with_all_params()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| 44606dd3_cb17_038d_86d1_0dd536e91d42
  9adc0698_1714_3e8c_79a9_63ed9ebccf5d["test_raw_response_list()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| 9adc0698_1714_3e8c_79a9_63ed9ebccf5d
  16c0f74a_7291_a6b0_e03d_766dc669d1ba["test_streaming_response_list()"]
  3599ed31_8a6e_9753_ec3c_978d9ca4f1b5 -->|method| 16c0f74a_7291_a6b0_e03d_766dc669d1ba

Relationship Graph

Source Code

tests/api_resources/messages/test_batches.py lines 278–532

class TestAsyncBatches:
    parametrize = pytest.mark.parametrize(
        "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
    )

    @parametrize
    async def test_method_create(self, async_client: AsyncAnthropic) -> None:
        batch = await async_client.messages.batches.create(
            requests=[
                {
                    "custom_id": "my-custom-id-1",
                    "params": {
                        "max_tokens": 1024,
                        "messages": [
                            {
                                "content": "Hello, world",
                                "role": "user",
                            }
                        ],
                        "model": "claude-opus-4-6",
                    },
                }
            ],
        )
        assert_matches_type(MessageBatch, batch, path=["response"])

    @parametrize
    async def test_raw_response_create(self, async_client: AsyncAnthropic) -> None:
        response = await async_client.messages.batches.with_raw_response.create(
            requests=[
                {
                    "custom_id": "my-custom-id-1",
                    "params": {
                        "max_tokens": 1024,
                        "messages": [
                            {
                                "content": "Hello, world",
                                "role": "user",
                            }
                        ],
                        "model": "claude-opus-4-6",
                    },
                }
            ],
        )

        assert response.is_closed is True
        assert response.http_request.headers.get("X-Stainless-Lang") == "python"
        batch = response.parse()
        assert_matches_type(MessageBatch, batch, path=["response"])

    @parametrize
    async def test_streaming_response_create(self, async_client: AsyncAnthropic) -> None:
        async with async_client.messages.batches.with_streaming_response.create(
            requests=[
                {
                    "custom_id": "my-custom-id-1",
                    "params": {
                        "max_tokens": 1024,
                        "messages": [
                            {
                                "content": "Hello, world",
                                "role": "user",
                            }
                        ],
                        "model": "claude-opus-4-6",
                    },
                }
            ],
        ) as response:
            assert not response.is_closed
            assert response.http_request.headers.get("X-Stainless-Lang") == "python"

            batch = await response.parse()
            assert_matches_type(MessageBatch, batch, path=["response"])

        assert cast(Any, response.is_closed) is True

    @parametrize
    async def test_method_retrieve(self, async_client: AsyncAnthropic) -> None:
        batch = await async_client.messages.batches.retrieve(