Streaming

rboto exposes byte streams and modeled event streams as async iterators.

Byte streams

Read an S3 body in chunks:

import asyncio
from pathlib import Path

from rboto import S3Client


async def download(client: S3Client, bucket: str, key: str, destination: Path) -> None:
    response = await client.get_object(bucket=bucket, key=key)
    with destination.open("wb") as output:
        async for chunk in response.body:
            await asyncio.to_thread(output.write, chunk)

The file writes run in a worker thread. Each write finishes before the next chunk is requested. A failed download may leave a partial destination file.

For a body that fits in memory:

from rboto import S3Client


async def read_object(client: S3Client, bucket: str, key: str) -> bytes:
    response = await client.get_object(bucket=bucket, key=key)
    return await response.body.read()

.read() collects the remaining bytes. Streams cannot be rewound. Each response owns its stream; give each stream one consumer.

put_object(body=...) accepts bytes. Multipart operations are individual service methods; the Python API does not include a high-level transfer manager.

Event streams

Event-stream responses are typed mappings containing receivers:

from rboto import CloudWatchLogsClient


async def tail_logs(client: CloudWatchLogsClient, group_arn: str) -> None:
    response = await client.start_live_tail(log_group_identifiers=[group_arn])
    async for event in response["response_stream"]:
        print(event)

Inspect the tagged event variant your application needs. Iteration can raise an error after the initial call has succeeded, so include iteration in error handling.

Bidirectional streams

Pass a typed AsyncIterator of input events. The SDK pulls input events as it sends them; the returned iterator supplies output events.

from collections.abc import AsyncIterator

from rboto import BedrockRuntimeClient
from rboto_bedrock_runtime.types import InvokeModelWithBidirectionalStreamInput


async def stream(
    client: BedrockRuntimeClient,
    model_id: str,
    events: AsyncIterator[InvokeModelWithBidirectionalStreamInput],
) -> None:
    response = await client.invoke_model_with_bidirectional_stream(model_id=model_id, body=events)
    async for event in response["body"]:
        print(event)

This helper forwards already encoded events. Input bytes and output interpretation depend on the selected model's protocol.