DynamoDB

Import DynamoDBClient from rboto. Install the service package following Getting started.

Put and get an item

This example needs a table with a string partition key named pk and no sort key.

from rboto import DynamoDBClient
from rboto_dynamodb.types import AttributeValue


async def put_and_get(client: DynamoDBClient, table: str) -> None:
    item: dict[str, AttributeValue] = {
        "pk": {"s": "USER#1"},
        "name": {"s": "Ada"},
        "age": {"n": "37"},
        "active": {"bool": True},
    }
    await client.put_item(table_name=table, item=item)
    response = await client.get_item(
        table_name=table,
        key={"pk": {"s": "USER#1"}},
        consistent_read=True,
    )
    print(response.item)

AttributeValue selects a type explicitly: s for strings, n for numbers, bool for booleans, b for bytes, m for maps, and l for lists. Numbers are represented as strings. rboto does not infer these types from arbitrary values.

put_item() replaces an existing item with the same key unless a condition rejects the write. A missing item gives response.item is None.

Query every page

from collections.abc import AsyncIterator

from rboto import DynamoDBClient
from rboto_dynamodb.types import AttributeValueOutput


async def query_items(
    client: DynamoDBClient,
    table: str,
    partition: str,
) -> AsyncIterator[dict[str, AttributeValueOutput]]:
    async for page in client.paginators.query(
        table_name=table,
        key_condition_expression="pk = :pk",
        expression_attribute_values={":pk": {"s": partition}},
    ):
        for item in page.items or []:
            yield item

The helper queries a table whose partition key is pk. The paginator follows last_evaluated_key even when a page contains no items: filtering can remove all items from a page. See Pagination for page size and stopping.

Conditional writes can raise ConditionalCheckFailedError. Batch writes require checking unprocessed_items; see Errors.