CRUD Operations¶

put¶

Insert or overwrite an item. If an item with the same primary key already exists, it is replaced.

await db.put(User(user_id="u1", name="Alice", email="alice@example.com"))

Conditional put¶

Guard the write with a condition expression. The put fails (raises ConditionalCheckFailedException) if the condition is not met.

from boto3.dynamodb.conditions import Attr

# Only insert if the item does not already exist
await db.put(
    User(user_id="u1", name="Alice"),
    condition_expression=Attr("user_id").not_exists(),
)

Signature:

async def put(
    self,
    item: DynamoModel,
    *,
    condition_expression: ConditionBase | None = None,
) -> None

get¶

Fetch a single item by its primary key. Returns None if not found.

user = await db.get(User, hash_key="u1")
if user is None:
    print("not found")

With a composite key:

order = await db.get(Order, hash_key="o1", range_key="2026-01-01T00:00:00")

Projection¶

Fetch only specific fields using ProjectionAttr:

from aiodynamodb import ProjectionAttr

user = await db.get(
    User,
    hash_key="u1",
    projection_expression=[ProjectionAttr("user_id"), ProjectionAttr("name")],
)

See Projections for details.

Consistent reads¶

user = await db.get(User, hash_key="u1", consistent_reads=True)

Signature:

async def get(
    self,
    model: type[T],
    *,
    hash_key: KeyT,
    range_key: KeyT | None = None,
    consistent_reads: bool = False,
    projection_expression: list[ProjectionAttr] | None = None,
) -> T | None

delete¶

Delete an item. Pass a model instance — the client extracts the key from it.

await db.delete(User(user_id="u1", name="Alice"))

Conditional delete¶

from boto3.dynamodb.conditions import Attr

await db.delete(
    User(user_id="u1", name="Alice"),
    condition_expression=Attr("user_id").exists(),
)

Signature:

async def delete(
    self,
    item: DynamoModel,
    *,
    condition_expression: ConditionBase | None = None,
) -> None

Exception handling¶

Condition expression failures raise ConditionalCheckFailedException. Access exception classes via db.exceptions():

ex = await db.exceptions()

try:
    await db.put(
        User(user_id="u1", name="Alice"),
        condition_expression=Attr("user_id").not_exists(),
    )
except ex.ConditionalCheckFailedException:
    print("item already exists")

See Exceptions for the full error handling guide.