Logging

Call configure_logging() once to enable JSON logs for every rboto client. No handler, formatter, or interceptor registration is required.

from rboto import S3Client, configure_logging

configure_logging()
client = S3Client()
await client.list_buckets()

Logs go to stdout. Each service has its own logger name: rboto.s3, rboto.dynamodb, rboto.bedrock-runtime, and so on. The name is also available as client.logger_name.

Every operation log includes a UTC timestamp, level, logger, service, operation, message, and operation metadata. JSON is formatted automatically, including when stdout is collected by a container or Lambda.

Set levels and disable services

Use one configuration for the whole application:

configure_logging(
    level="INFO",
    service_levels={"s3": "OFF", "dynamodb": "DEBUG"},
)

This disables S3 logs, enables DynamoDB retry diagnostics, and uses INFO for the other services. Settings apply to existing clients, future clients, Session-based clients, and compatibility factories.

Calling configure_logging() again replaces the default level and per-service overrides. It resets rboto service loggers to that configuration and reuses its handler; it does not add duplicate handlers. The application root logger and its handlers are unchanged. rboto records do not propagate to the root handler.

To disable all operation logging:

configure_logging(level="OFF")

To enable only S3:

configure_logging(level="OFF", service_levels={"s3": "INFO"})

For a change to one logger without reconfiguring the others, use get_logger(). It returns the standard Python logger with the service's name:

import asyncio
import logging

from rboto import S3Client, configure_logging, get_logger

client = S3Client()
configure_logging()  # Also applies to clients created before this call.


async def main() -> None:
    logger = get_logger("s3")
    assert logger.name == client.logger_name == "rboto.s3"

    await client.list_buckets()
    logger.disabled = True
    await client.list_buckets()
    logger.disabled = False
    logger.setLevel(logging.DEBUG)
    await client.list_buckets()


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

get_logger("s3") and logging.getLogger("rboto.s3") return the same object. Disabling that logger skips automatic logging and its native metadata collection. Custom application middleware, if configured, still runs.

No operation logs are produced until configure_logging() is called. Without logging or application middleware, operations use direct native dispatch.

What gets logged

There is one completion record per invocation when its level is enabled. A retry record summarizes up to 32 completed attempts after native dispatch finishes.

Logging automatically wraps application middleware, so cache hits, policy blocks, and error recovery are included. A cache hit has dispatched=false and zero native attempts. A policy failure is an error even if no request was sent. If middleware recovers from a native error, the final outcome is successful while the native error fields retain the underlying failure.

For streaming operations, completion means the response is available. Logging does not read the stream or measure later consumption; subsequent stream errors remain with the code consuming it.

JSON fields

Unavailable values are null. Trace and retry detail fields are omitted when not applicable. Attempt duration excludes retry backoff. Cancellation can prevent the native collector from delivering attempt counts, so unknown counts remain null.

Automatic operation logs never inspect parameters, response properties, streams, to_dict(), exception messages, or tracebacks. Correlation and trace IDs are supplied by your application; use IDs appropriate for logs.

Concurrent calls and correlation

Each task keeps its own operation and correlation IDs:

import asyncio

from rboto import DynamoDBClient, S3Client, configure_logging, operation_context

# One configuration for every client: JSON to stdout, no formatter required.
configure_logging(level="INFO", service_levels={"dynamodb": "DEBUG", "s3": "OFF"})


async def main() -> None:
    dynamo = DynamoDBClient()
    storage = S3Client()

    async def tables() -> None:
        with operation_context(correlation_id="list-tables"):
            await dynamo.list_tables()

    async def buckets() -> None:
        with operation_context(correlation_id="list-buckets"):
            await storage.list_buckets()

    async with asyncio.TaskGroup() as group:
        group.create_task(tables())
        group.create_task(buckets())


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

Trace IDs

An optional synchronous provider can return LogTraceContext or None. The provider runs on the operation's task when the record's level is enabled. It does not require a telemetry dependency:

import asyncio
from contextvars import ContextVar

from rboto import DynamoDBClient, LogTraceContext, configure_logging

trace_context: ContextVar[LogTraceContext | None] = ContextVar("application_trace", default=None)
configure_logging(trace_context=trace_context.get)


async def main() -> None:
    client = DynamoDBClient()
    # In an application, obtain these IDs from the tracer's current span.
    token = trace_context.set(
        LogTraceContext(
            trace_id="0123456789abcdef0123456789abcdef",
            span_id="0123456789abcdef",
        )
    )
    try:
        await client.list_tables()
    finally:
        trace_context.reset(token)


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

Use your tracer's active span IDs in the provider. No spans are created by logging.

AWS Lambda

Configure logging once, outside the handler. Attach the invocation ID with operation_context() and keep the client available for reuse:

import asyncio
from typing import Protocol

from rboto import DynamoDBClient, configure_logging, operation_context

configure_logging()
client = DynamoDBClient()


class LambdaContext(Protocol):
    @property
    def aws_request_id(self) -> str: ...


async def list_tables() -> dict[str, object]:
    response = await client.list_tables()
    return {"tables": response.table_names}


def lambda_handler(event: dict[str, object], context: LambdaContext) -> dict[str, object]:
    with operation_context(correlation_id=context.aws_request_id):
        return asyncio.run(list_tables())

The output is already JSON; no runtime formatter is needed. If you enable Lambda log-level filtering, choose a level that retains the rboto records you want. Context is restored on exit, including failures, so reused environments do not retain a previous invocation's correlation ID. See AWS Lambda for packaging and client reuse.

Output destination

Stdout is the default. Pass a text stream when the application needs a different destination, such as a file it manages:

with open("operations.jsonl", "a", encoding="utf-8") as output:
    configure_logging(stream=output)
    await client.list_buckets()
    configure_logging(level="OFF")

rboto does not close application-owned streams. Writes are synchronous; a slow destination can delay the caller. Ordinary logging failures do not replace the operation result or its original exception. Delivery is not retried, so use these records for diagnostics rather than guaranteed audit delivery.