Models

Models define the structure of your DynamoDB items and provide CRUD operations.

Key features

Typed attributes with defaults
Hash key and range key support
Required fields with required=True
Save, get, update, delete operations
Convert to/from dict

Getting started

Basic model

basic_model.py

from pydynox import Model, ModelConfig
from pydynox.attributes import BooleanAttribute, NumberAttribute, StringAttribute


class User(Model):
    model_config = ModelConfig(table="users")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()
    age = NumberAttribute(default=0)
    active = BooleanAttribute(default=True)

Tip

Want to see all supported attribute types? Check out the Attribute types guide.

Keys

Every model needs at least a hash key (partition key):

class User(Model):
    model_config = ModelConfig(table="users")

    pk = StringAttribute(partition_key=True)  # Required

Add a range key (sort key) for composite keys:

class User(Model):
    model_config = ModelConfig(table="users")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)  # Optional

Defaults and required fields

with_defaults.py

from pydynox import Model, ModelConfig
from pydynox.attributes import (
    BooleanAttribute,
    ListAttribute,
    MapAttribute,
    NumberAttribute,
    StringAttribute,
)


class User(Model):
    model_config = ModelConfig(table="users")

    pk = StringAttribute(partition_key=True)
    email = StringAttribute(required=True)  # Required field
    name = StringAttribute(default="")
    age = NumberAttribute(default=0)
    active = BooleanAttribute(default=True)
    tags = ListAttribute(default=[])
    settings = MapAttribute(default={})

CRUD operations

pydynox uses an async-first API. Methods without prefix are async (default), methods with sync_ prefix are sync.

Async (default)Sync (use sync_ prefix)

"""Example: CRUD operations (async - default)."""

import asyncio

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute


class User(Model):
    model_config = ModelConfig(table="users")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()
    age = NumberAttribute(default=0)


async def main():
    # Create
    user = User(pk="USER#123", sk="PROFILE", name="John", age=30)
    await user.save()

    # Read
    user = await User.get(pk="USER#123", sk="PROFILE")
    if user:
        print(user.name)  # John

    # Update - full
    user.name = "Jane"
    await user.save()

    # Update - partial
    await user.update(name="Jane", age=31)

    # Delete
    await user.delete()


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

"""Example: CRUD operations (sync - use sync_ prefix)."""

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute


class User(Model):
    model_config = ModelConfig(table="users")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()
    age = NumberAttribute(default=0)


# Create
user = User(pk="USER#123", sk="PROFILE", name="John", age=30)
user.sync_save()

# Read
user = User.sync_get(pk="USER#123", sk="PROFILE")
if user:
    print(user.name)  # John

# Update - full
user.name = "Jane"
user.sync_save()

# Update - partial
user.sync_update(name="Jane", age=31)

# Delete
user.sync_delete()

Create

To create a new item, instantiate your model and call save():

user = User(pk="USER#123", sk="PROFILE", name="John", age=30)
await user.save()  # async
user.sync_save()   # sync

If an item with the same key already exists, save() replaces it completely. This is how DynamoDB works - there's no separate "create" vs "update" at the API level.

Read

To get an item by its key, use the class method get():

# Async
user = await User.get(pk="USER#123", sk="PROFILE")
if user:
    print(user.name)
else:
    print("User not found")

# Sync
user = User.sync_get(pk="USER#123", sk="PROFILE")

get() returns None if the item doesn't exist. Always check for None before using the result.

If your table has only a hash key (no range key), you only need to pass the hash key:

user = await User.get(pk="USER#123")

Consistent reads

By default, get() uses eventually consistent reads. For strongly consistent reads, use consistent_read=True:

Async (default)Sync (use sync_ prefix)

"""Consistent read examples (async - default)."""

import asyncio

from pydynox import DynamoDBClient, Model, ModelConfig
from pydynox.attributes import StringAttribute

client = DynamoDBClient(region="us-east-1")


# Option 1: Per-operation (highest priority)
class User(Model):
    model_config = ModelConfig(table="users", client=client)

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()


# Option 2: Model-level default
class Order(Model):
    model_config = ModelConfig(
        table="orders",
        client=client,
        consistent_read=True,  # All reads are strongly consistent
    )

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)


async def main():
    # Eventually consistent (default)
    _user = await User.get(pk="USER#123", sk="PROFILE")

    # Strongly consistent
    _user = await User.get(pk="USER#123", sk="PROFILE", consistent_read=True)

    # Uses strongly consistent read (from model_config)
    _order = await Order.get(pk="ORDER#456", sk="ITEM#1")

    # Override to eventually consistent for this call
    _order = await Order.get(pk="ORDER#456", sk="ITEM#1", consistent_read=False)


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

"""Consistent read examples (sync - use sync_ prefix)."""

from pydynox import DynamoDBClient, Model, ModelConfig
from pydynox.attributes import StringAttribute

client = DynamoDBClient(region="us-east-1")


# Option 1: Per-operation (highest priority)
class User(Model):
    model_config = ModelConfig(table="users", client=client)

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()


# Eventually consistent (default)
user = User.sync_get(pk="USER#123", sk="PROFILE")

# Strongly consistent
user = User.sync_get(pk="USER#123", sk="PROFILE", consistent_read=True)


# Option 2: Model-level default
class Order(Model):
    model_config = ModelConfig(
        table="orders",
        client=client,
        consistent_read=True,  # All reads are strongly consistent
    )

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)


# Uses strongly consistent read (from model_config)
order = Order.sync_get(pk="ORDER#456", sk="ITEM#1")

# Override to eventually consistent for this call
order = Order.sync_get(pk="ORDER#456", sk="ITEM#1", consistent_read=False)

When to use strongly consistent reads:

You need to read data right after writing it
Your app can't tolerate stale data (even for a second)
You're building a financial or inventory system

Trade-offs:

	Eventually consistent	Strongly consistent
Latency	Lower	Higher
Cost	0.5 RCU per 4KB	1 RCU per 4KB
Availability	Higher	Lower during outages

Most apps work fine with eventually consistent reads. Use strongly consistent only when you need it.

Update

There are two ways to update an item:

Full update with save(): Change attributes and call save(). This replaces the entire item:

user = await User.get(pk="USER#123", sk="PROFILE")
user.name = "Jane"
user.age = 31
await user.save()

# Sync
user = User.sync_get(pk="USER#123", sk="PROFILE")
user.name = "Jane"
user.sync_save()

Partial update with update(): Update specific fields without touching others:

user = await User.get(pk="USER#123", sk="PROFILE")
await user.update(name="Jane", age=31)

# Sync
user.sync_update(name="Jane", age=31)

The difference matters when you have many attributes. With save(), you send all attributes to DynamoDB. With update(), you only send the changed ones.

update() also updates the local object, so user.name is "Jane" after the call.

Delete

To delete an item, call delete() on an instance:

user = await User.get(pk="USER#123", sk="PROFILE")
await user.delete()

# Sync
user.sync_delete()

After deletion, the object still exists in Python, but the item is gone from DynamoDB.

Update and delete by key

Sometimes you want to update or delete an item without fetching it first. The traditional approach requires two DynamoDB calls:

user = await User.get(pk="USER#123", sk="PROFILE")  # Call 1
await user.update(name="Jane")                       # Call 2

Use update_by_key() and delete_by_key() to do it in one call:

Async (default)Sync (use sync_ prefix)

"""Example: update_by_key and delete_by_key operations (async - default)."""

import asyncio

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute


class User(Model):
    model_config = ModelConfig(table="users")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()
    age = NumberAttribute()


async def main():
    # Update without fetching first - single DynamoDB call
    await User.update_by_key(pk="USER#123", sk="PROFILE", name="Jane", age=31)

    # Delete without fetching first - single DynamoDB call
    await User.delete_by_key(pk="USER#123", sk="PROFILE")

    # Compare with traditional approach (2 calls):
    # user = await User.get(pk="USER#123", sk="PROFILE")  # Call 1
    # await user.update(name="Jane")                       # Call 2


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

"""Example: update_by_key and delete_by_key operations (sync - use sync_ prefix)."""

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute


class User(Model):
    model_config = ModelConfig(table="users")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()
    age = NumberAttribute()


# Update without fetching first - single DynamoDB call
User.sync_update_by_key(pk="USER#123", sk="PROFILE", name="Jane", age=31)

# Delete without fetching first - single DynamoDB call
User.sync_delete_by_key(pk="USER#123", sk="PROFILE")

# Compare with traditional approach (2 calls):
# user = User.sync_get(pk="USER#123", sk="PROFILE")  # Call 1
# user.sync_update(name="Jane")                       # Call 2

This is about 2x faster because you skip the read operation.

When to use:

Bulk updates where you know the keys
Background jobs that process many items
Any case where you don't need the current item data

Trade-offs:

Method	DynamoDB calls	Returns item?	Runs hooks?
`get()` + `update()`	2	Yes	Yes
`update_by_key()`	1	No	No
`get()` + `delete()`	2	Yes	Yes
`delete_by_key()`	1	No	No

Note

These methods don't run lifecycle hooks. If you need hooks, use the traditional get() + update()/delete() approach.

Advanced

ModelConfig options

Option	Type	Default	Description
`table`	str	Required	DynamoDB table name
`client`	DynamoDBClient	None	Client to use (uses default if None)
`skip_hooks`	bool	False	Skip lifecycle hooks
`max_size`	int	None	Max item size in bytes
`consistent_read`	bool	False	Use strongly consistent reads by default

Setting a default client

Instead of passing a client to each model, set a default client once:

from pydynox import DynamoDBClient, set_default_client

# At app startup
client = DynamoDBClient(region="us-east-1", profile="prod")
set_default_client(client)

# All models use this client
class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)

class Order(Model):
    model_config = ModelConfig(table="orders")
    pk = StringAttribute(partition_key=True)

Override client per model

Use a different client for specific models:

# Default client for most models
set_default_client(prod_client)

# Special client for audit logs
audit_client = DynamoDBClient(region="eu-west-1")

class AuditLog(Model):
    model_config = ModelConfig(
        table="audit_logs",
        client=audit_client,  # Uses different client
    )
    pk = StringAttribute(partition_key=True)

Converting to dict

user = User(pk="USER#123", sk="PROFILE", name="John")
data = user.to_dict()
# {'pk': 'USER#123', 'sk': 'PROFILE', 'name': 'John'}

Creating from dict

data = {'pk': 'USER#123', 'sk': 'PROFILE', 'name': 'John'}
user = User.from_dict(data)

Skipping hooks

If you have lifecycle hooks but want to skip them for a specific operation:

user.save(skip_hooks=True)
user.delete(skip_hooks=True)
user.update(skip_hooks=True, name="Jane")

This is useful for:

Data migrations where validation might fail on old data
Bulk operations where you want maximum speed
Fixing bad data that wouldn't pass validation

You can also disable hooks for all operations on a model:

class User(Model):
    model_config = ModelConfig(table="users", skip_hooks=True)

Warning

Be careful when skipping hooks. If you have validation in before_save, skipping it means invalid data can be saved to DynamoDB.

Error handling

DynamoDB operations can fail for various reasons. Common errors:

Exception	Cause
`ResourceNotFoundException`	Table doesn't exist
`ProvisionedThroughputExceededException`	Exceeded capacity (throttled)
`ValidationException`	Invalid data (item too large, bad key, etc.)
`ConditionalCheckFailedException`	Conditional write failed
`ItemTooLargeException`	Item exceeds `max_size` (Python-only, before DynamoDB call)

Wrap operations in try/except:

from pydynox.exceptions import (
    ResourceNotFoundException,
    ProvisionedThroughputExceededException,
    PydynoxException,
)

try:
    await user.save()
except ResourceNotFoundException:
    print("Table doesn't exist")
except ProvisionedThroughputExceededException:
    print("Rate limited, try again")
except PydynoxException as e:
    print(f"DynamoDB error: {e}")

See Exceptions for the full list.

Testing your code

Test models without DynamoDB using the built-in memory backend:

testing_models.py

"""Testing models with pydynox_memory_backend (sync tests use sync_ prefix)."""

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()
    age = NumberAttribute(default=0)


def test_create_and_get(pydynox_memory_backend):
    """Test creating and getting a model."""
    user = User(pk="USER#1", sk="PROFILE", name="John", age=30)
    user.sync_save()

    found = User.sync_get(pk="USER#1", sk="PROFILE")
    assert found is not None
    assert found.name == "John"
    assert found.age == 30


def test_update(pydynox_memory_backend):
    """Test updating a model."""
    user = User(pk="USER#1", sk="PROFILE", name="John")
    user.sync_save()

    user.sync_update(name="Jane", age=25)

    found = User.sync_get(pk="USER#1", sk="PROFILE")
    assert found.name == "Jane"
    assert found.age == 25


def test_delete(pydynox_memory_backend):
    """Test deleting a model."""
    user = User(pk="USER#1", sk="PROFILE", name="John")
    user.sync_save()

    user.sync_delete()

    assert User.sync_get(pk="USER#1", sk="PROFILE") is None


def test_get_not_found(pydynox_memory_backend):
    """Test getting a non-existent model."""
    found = User.sync_get(pk="USER#999", sk="PROFILE")
    assert found is None


def test_update_by_key(pydynox_memory_backend):
    """Test updating by key without fetching first."""
    User(pk="USER#1", sk="PROFILE", name="John").sync_save()

    User.sync_update_by_key(pk="USER#1", sk="PROFILE", name="Jane")

    found = User.sync_get(pk="USER#1", sk="PROFILE")
    assert found.name == "Jane"


def test_delete_by_key(pydynox_memory_backend):
    """Test deleting by key without fetching first."""
    User(pk="USER#1", sk="PROFILE", name="John").sync_save()

    User.sync_delete_by_key(pk="USER#1", sk="PROFILE")

    assert User.sync_get(pk="USER#1", sk="PROFILE") is None

No setup needed. Just add pydynox_memory_backend to your test function. See Testing for more details.

Next steps

Query - Query items by hash key with conditions
Attribute types - All available attribute types
Indexes - Query by non-key attributes with GSIs
Conditions - Conditional writes
Hooks - Lifecycle hooks for validation