Async-only

Every rboto service operation is an async def method. There are no synchronous operation variants or sync_ methods.

While a request waits for the network, the Python event loop can run other coroutines. Sequential await calls still run one at a time. Your application chooses which requests can run concurrently.

Scripts

Put requests in an async function and start the event loop once:

import asyncio

from rboto import STSClient


async def main() -> None:
    client = STSClient(region="us-east-1")
    response = await client.get_caller_identity()
    print(response.account)
    print(response.arn)


if __name__ == "__main__":
    asyncio.run(main())

Constructing the client does not require await. Calling an operation without awaiting or scheduling it creates a coroutine and does not send the request.

Applications and notebooks

Inside an async handler, job, or notebook with an active event loop, await the operation directly. Do not call asyncio.run() inside an already running loop.

from rboto import SSMClient


async def read_parameter(client: SSMClient, name: str) -> str | None:
    response = await client.get_parameter(name=name, with_decryption=True)
    if response.parameter is None:
        return None
    return response.parameter.value

Create clients during application setup and pass them to code that needs them. The current clients do not implement context managers or a close() method.

For a synchronous Lambda entry point that calls async code, see AWS Lambda.

Concurrent requests

Use asyncio.gather() for a small, known set of independent operations:

import asyncio

from rboto import SSMClient


async def read_settings(client: SSMClient) -> None:
    database, queue = await asyncio.gather(
        client.get_parameter(name="/app/database"),
        client.get_parameter(name="/app/queue"),
    )
    for response in (database, queue):
        if response.parameter is not None:
            print(response.parameter.name, response.parameter.value)

Both calls reuse one client. Concurrency does not change service quotas or make an individual request inherently faster.

For large inputs, keep a bounded number of tasks in flight. A semaphore limits active calls, but creating one task for every input still consumes memory. Use bounded batches or a fixed set of workers.

Deadlines

Use asyncio.timeout() to bound your coroutine's wait:

import asyncio

from rboto import STSClient


async def identify(client: STSClient) -> str | None:
    async with asyncio.timeout(5):
        response = await client.get_caller_identity()
        return response.arn

For streams, keep iteration inside the timeout block if the deadline must include reading the body. Cancellation does not prove that a remote write was rejected. Use the operation's idempotency token when available and account for an uncertain result before repeating a write.