Skip to content

Smart save

By default, pydynox tracks which fields changed and only sends those to DynamoDB.

Why this matters

  • Atomic partial updates (no race conditions from read-modify-write)
  • Conditional updates (built-in support)
  • Network bandwidth savings (sending less data over the wire)

Note that it doesn't save on WCUs as DynamoDB still charges based on the size of the item, not the size of attributes that were updated.

How it works

When you load an item from DynamoDB, pydynox stores a snapshot of the original values. When you call save(), it compares current values with the original and only sends the changed fields using UpdateItem.

"""Basic smart save example - only changed fields are sent to DynamoDB."""

import asyncio

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


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()
    email = StringAttribute()
    bio = StringAttribute()


async def main():
    # Load a 4KB item with 20 fields
    user = await User.get(pk="USER#1", sk="PROFILE")
    if user:
        # Change one field
        user.name = "New Name"

        # Only sends 'name' to DynamoDB (not all 4KB)
        await user.save()


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

Check if item changed

Use is_dirty and changed_fields to see what changed:

"""Check if item changed using is_dirty and changed_fields."""

import asyncio

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


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()
    email = StringAttribute()


async def main():
    user = await User.get(pk="USER#1", sk="PROFILE")
    if user:
        print(user.is_dirty)  # False

        user.name = "New Name"
        print(user.is_dirty)  # True
        print(user.changed_fields)  # ["name"]

        user.email = "new@example.com"
        print(user.changed_fields)  # ["name", "email"]


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

Force full replace

If you need to replace the entire item (using PutItem instead of UpdateItem), use full_replace=True:

"""Force full replace using PutItem instead of UpdateItem."""

import asyncio

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


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


async def main():
    user = await User.get(pk="USER#1", sk="PROFILE")
    if user:
        user.name = "New Name"

        # Forces PutItem with all fields
        await user.save(full_replace=True)


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

Use this when:

  • You want to remove fields that are not in the model
  • You need PutItem behavior for some reason

New items

New items (not loaded from DynamoDB) always use PutItem:

"""New items always use PutItem, then smart save kicks in."""

import asyncio

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


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


async def main():
    # New item - uses PutItem
    user = User(pk="USER#new", sk="PROFILE", name="John")
    await user.save()  # PutItem

    # After save, tracking is enabled
    user.name = "Jane"
    await user.save()  # UpdateItem (smart save)

    # Cleanup
    await user.delete()


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

With conditions

Smart save works with conditions:

"""Smart save works with conditions."""

import asyncio

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


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    status = StringAttribute()


async def main():
    user = await User.get(pk="USER#1", sk="PROFILE")
    if user:
        user.status = "active"

        # UpdateItem with condition
        try:
            await user.save(condition=User.status == "pending")
        except Exception:
            # Condition failed - status was not "pending"
            pass


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

With optimistic locking

Smart save works with version attributes:

"""Smart save works with optimistic locking (version attribute)."""

import asyncio

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


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    name = StringAttribute()
    version = VersionAttribute()


async def main():
    user = await User.get(pk="USER#1", sk="PROFILE")
    if user:
        user.name = "New Name"

        # UpdateItem with version check
        await user.save()


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