Repository: pydantic/pydantic-ai
File: pydantic_ai_slim/pydantic_ai/models/bedrock.py (Python)

Code before:
                            yield maybe_event
                case _:
                    pass  # pyright wants match statements to be exhaustive

    @property
    def model_name(self) -> str:
        """Get the model name of the response."""
        return self._model_name

    @property
    def provider_name(self) -> str:
        """Get the provider name."""
        return self._provider_name

    @property
    def provider_url(self) -> str:
        """Get the provider base URL."""
        return self._provider_url

    @property
    def timestamp(self) -> datetime:
        return self._timestamp

    def _map_usage(self, metadata: ConverseStreamMetadataEventTypeDef) -> usage.RequestUsage:
        input_tokens = metadata['usage'].get('inputTokens', 0)
        output_tokens = metadata['usage'].get('outputTokens', 0)
        cache_read_tokens = metadata['usage'].get('cacheReadInputTokens', 0)
        cache_write_tokens = metadata['usage'].get('cacheWriteInputTokens', 0)
        return usage.RequestUsage(
            input_tokens=input_tokens + cache_write_tokens + cache_read_tokens,
            output_tokens=output_tokens,
            cache_read_tokens=cache_read_tokens,
            cache_write_tokens=cache_write_tokens,
        )


class _AsyncIteratorWrapper(Generic[T]):
    """Wrap a synchronous iterator in an async iterator."""

    def __init__(self, sync_iterator: Iterable[T]):
        self.sync_iterator = iter(sync_iterator)

    def __aiter__(self):
        return self

    async def __anext__(self) -> T:
        try:
            return await anyio.to_thread.run_sync(next, self.sync_iterator)
        except RuntimeError as e:
            if type(e.__cause__) is StopIteration:
                raise StopAsyncIteration

Review comment (suggestion):
if this was `metadata['usage']['inputTokens']` lets keep it like that now as well, the difference between `[]` and `get` is a hard assumption in bedrock's API's response struture

Code after:
                            yield maybe_event
                case _:
                    pass  # pyright wants match statements to be exhaustive

    @property
    def model_name(self) -> str:
        """Get the model name of the response."""
        return self._model_name

    @property
    def provider_name(self) -> str:
        """Get the provider name."""
        return self._provider_name

    @property
    def provider_url(self) -> str:
        """Get the provider base URL."""
        return self._provider_url

    @property
    def timestamp(self) -> datetime:
        return self._timestamp

    def _map_usage(self, metadata: ConverseStreamMetadataEventTypeDef) -> usage.RequestUsage:
        input_tokens = metadata['usage']['inputTokens']
        output_tokens = metadata['usage']['outputTokens']
        cache_read_tokens = metadata['usage'].get('cacheReadInputTokens', 0)
        cache_write_tokens = metadata['usage'].get('cacheWriteInputTokens', 0)
        return usage.RequestUsage(
            input_tokens=input_tokens + cache_write_tokens + cache_read_tokens,
            output_tokens=output_tokens,
            cache_read_tokens=cache_read_tokens,
            cache_write_tokens=cache_write_tokens,
        )


class _AsyncIteratorWrapper(Generic[T]):
    """Wrap a synchronous iterator in an async iterator."""

    def __init__(self, sync_iterator: Iterable[T]):
        self.sync_iterator = iter(sync_iterator)

    def __aiter__(self):
        return self

    async def __anext__(self) -> T:
        try:
            return await anyio.to_thread.run_sync(next, self.sync_iterator)
        except RuntimeError as e:
            if type(e.__cause__) is StopIteration:
                raise StopAsyncIteration