Atomic updates

Atomic updates modify values in DynamoDB without reading them first. This avoids race conditions when multiple requests update the same item.

Key features

Increment/decrement numbers without read-modify-write
Append/prepend items to lists
Set values only if attribute doesn't exist
Remove attributes
Combine multiple operations in one request
Add conditions for safe updates

The problem with read-modify-write

Without atomic updates, incrementing a counter looks like this:

# Dangerous - race condition!
user = User.get(pk="USER#123")
user.login_count = user.login_count + 1
user.save()

If two requests run at the same time:

Request A reads login_count = 5
Request B reads login_count = 5
Request A writes login_count = 6
Request B writes login_count = 6

You lost an increment. The count should be 7, but it's 6.

Atomic updates solve this by doing the math in DynamoDB:

# Safe - no race condition
user.update(atomic=[User.login_count.add(1)])

Getting started

Counters

The most common use case. Increment or decrement a number:

counter.py

"""Atomic counter example."""

import asyncio

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


class PageView(Model):
    model_config = ModelConfig(table="analytics")

    pk = StringAttribute(partition_key=True)  # page_url
    sk = StringAttribute(sort_key=True)  # date
    views = NumberAttribute()


async def main():
    # Increment counter without reading first
    page = PageView(pk="/home", sk="2024-01-15", views=0)
    await page.save()

    # Each request increments atomically
    await page.update(atomic=[PageView.views.add(1)])

    # Multiple increments are safe - no race conditions
    # Request 1: views = 0 + 1 = 1
    # Request 2: views = 1 + 1 = 2
    # Request 3: views = 2 + 1 = 3


asyncio.run(main())

Each add(1) is atomic. Even with thousands of concurrent requests, every increment is counted.

add() starts from zero when the attribute is missing, so the same call works for the first increment and every one after it. This is useful for aggregates you build up over time:

aggregate_counters.py

"""Building up aggregates with add() on attributes that don't exist yet."""

import asyncio

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


class WalletAggregate(Model):
    model_config = ModelConfig(table="wallet-transaction-aggregates")

    wallet_id = StringAttribute(partition_key=True)
    currency = StringAttribute(sort_key=True)

    total_amount = NumberAttribute()
    transaction_count = NumberAttribute()


async def main():
    # The aggregate item does not exist yet - no save() needed
    aggregate = WalletAggregate(wallet_id="wallet-123", currency="USD")

    # add() starts from zero, so this works on the first call
    await aggregate.update(
        atomic=[
            WalletAggregate.total_amount.add(25),
            WalletAggregate.transaction_count.add(1),
        ]
    )
    # total_amount: 25, transaction_count: 1

    # The same code keeps incrementing from there
    await aggregate.update(
        atomic=[
            WalletAggregate.total_amount.add(10),
            WalletAggregate.transaction_count.add(1),
        ]
    )
    # total_amount: 35, transaction_count: 2


asyncio.run(main())

DynamoDB creates the item from the key when it is missing, so you don't need a read or a separate save first. That also means a wrong key writes a new item instead of failing. When you need the old behavior, add a condition so the update only touches an existing item:

await aggregate.update(
    atomic=[WalletAggregate.total_amount.add(25)],
    condition=WalletAggregate.wallet_id.exists(),
)

Safe balance transfer

Combine atomic updates with conditions to prevent overdrafts:

balance_transfer.py

"""Safe balance transfer with condition."""

import asyncio

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


class Account(Model):
    model_config = ModelConfig(table="accounts")

    pk = StringAttribute(partition_key=True)  # account_id
    balance = NumberAttribute()


async def withdraw(account: Account, amount: int) -> bool:
    """Withdraw money only if balance is sufficient."""
    try:
        await account.update(
            atomic=[Account.balance.add(-amount)],
            condition=Account.balance >= amount,
        )
        return True
    except ConditionalCheckFailedException:
        return False


async def main():
    # Usage
    account = Account(pk="ACC#123", balance=100)
    await account.save()

    # This succeeds - balance goes from 100 to 50
    success = await withdraw(account, 50)
    print(f"Withdrew 50: {success}")  # True

    # This fails - balance is 50, can't withdraw 75
    success = await withdraw(account, 75)
    print(f"Withdrew 75: {success}")  # False


asyncio.run(main())

The condition balance >= amount is checked atomically with the update. If the balance is too low, the whole operation fails.

Advanced

Inventory management

Track stock and reservations atomically:

inventory.py

"""Inventory management with atomic updates."""

import asyncio

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


class Product(Model):
    model_config = ModelConfig(table="products")

    pk = StringAttribute(partition_key=True)  # product_id
    stock = NumberAttribute()
    reserved = NumberAttribute()


class OutOfStock(Exception):
    pass


async def reserve_stock(product: Product, quantity: int) -> None:
    """Reserve stock for an order."""
    try:
        await product.update(
            atomic=[
                Product.stock.add(-quantity),
                Product.reserved.add(quantity),
            ],
            condition=Product.stock >= quantity,
        )
    except ConditionalCheckFailedException:
        raise OutOfStock(f"Not enough stock for {product.pk}")


async def release_stock(product: Product, quantity: int) -> None:
    """Release reserved stock (order cancelled)."""
    await product.update(
        atomic=[
            Product.stock.add(quantity),
            Product.reserved.add(-quantity),
        ]
    )


async def main():
    # Usage
    product = Product(pk="SKU#ABC123", stock=10, reserved=0)
    await product.save()

    # Reserve 3 units
    await reserve_stock(product, 3)
    # stock: 7, reserved: 3

    # Try to reserve 10 more - fails
    try:
        await reserve_stock(product, 10)
    except OutOfStock:
        print("Cannot reserve - not enough stock")

    # Cancel order - release the 3 units
    await release_stock(product, 3)
    # stock: 10, reserved: 0


asyncio.run(main())

Both stock and reserved are updated in one atomic operation. No item can be double-sold.

Rate limiting

Enforce API rate limits with atomic counters:

rate_limit_counter.py

"""API rate limiting with atomic counters."""

import asyncio

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


class ApiUsage(Model):
    model_config = ModelConfig(table="api_usage")

    pk = StringAttribute(partition_key=True)  # user_id
    sk = StringAttribute(sort_key=True)  # date (YYYY-MM-DD)
    requests = NumberAttribute()


class RateLimitExceeded(Exception):
    pass


async def track_request(user_id: str, date: str, daily_limit: int = 1000) -> int:
    """Track API request and enforce rate limit.

    Returns the new request count.
    Raises RateLimitExceeded if over limit.
    """
    usage = ApiUsage(pk=user_id, sk=date)

    try:
        # add() starts from zero, so the first request of the day needs no
        # special case. The condition also covers the missing attribute.
        await usage.update(
            atomic=[ApiUsage.requests.add(1)],
            condition=(ApiUsage.requests < daily_limit) | ApiUsage.requests.not_exists(),
        )
    except ConditionalCheckFailedException:
        raise RateLimitExceeded(f"User {user_id} exceeded {daily_limit} requests/day")

    # Consistent read, otherwise the count we just wrote may not be visible yet
    updated = await ApiUsage.get(pk=user_id, sk=date, consistent_read=True)
    return updated.requests


async def main():
    try:
        count = await track_request("user_123", "2024-01-15")
        print(f"Request #{count} recorded")
    except RateLimitExceeded as e:
        print(f"Rate limit hit: {e}")


asyncio.run(main())

The condition ensures you can't exceed the limit, even with concurrent requests.

Shopping cart

Manage cart items and totals:

shopping_cart.py

"""Shopping cart with list operations."""

import asyncio

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


class Cart(Model):
    model_config = ModelConfig(table="carts")

    pk = StringAttribute(partition_key=True)  # user_id
    items = ListAttribute()
    total = NumberAttribute()


async def add_to_cart(cart: Cart, item: dict, price: float) -> None:
    """Add item to cart and update total."""
    await cart.update(
        atomic=[
            Cart.items.append([item]),
            Cart.total.add(price),
        ]
    )


async def apply_discount(cart: Cart, discount: float) -> None:
    """Apply discount to cart total."""
    await cart.update(
        atomic=[Cart.total.add(-discount)],
        condition=Cart.total >= discount,
    )


async def main():
    # Usage
    cart = Cart(pk="USER#123", items=[], total=0)
    await cart.save()

    # Add items
    await add_to_cart(cart, {"sku": "SHIRT-M", "qty": 1}, 29.99)
    await add_to_cart(cart, {"sku": "PANTS-L", "qty": 2}, 49.99)

    # Cart now has:
    # items: [{"sku": "SHIRT-M", "qty": 1}, {"sku": "PANTS-L", "qty": 2}]
    # total: 79.98

    # Apply $10 discount
    await apply_discount(cart, 10.00)
    # total: 69.98


asyncio.run(main())

List operations

Add items to lists without reading the whole list:

user_tags.py

"""Managing user tags with list operations."""

import asyncio

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


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

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


async def main():
    # Create user with initial tags
    user = User(pk="USER#123", sk="PROFILE", tags=["member"])
    await user.save()

    # Add tags to the end
    await user.update(atomic=[User.tags.append(["premium", "verified"])])
    # tags: ["member", "premium", "verified"]

    # Add tags to the beginning
    await user.update(atomic=[User.tags.prepend(["vip"])])
    # tags: ["vip", "member", "premium", "verified"]


asyncio.run(main())

Default values

Set a value only if the attribute doesn't exist:

if_not_exists.py

"""Using if_not_exists for default values."""

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)
    login_count = NumberAttribute()
    score = NumberAttribute()


async def main():
    # User without login_count
    user = User(pk="USER#123", sk="PROFILE")
    await user.save()

    # Set default value only if attribute doesn't exist
    await user.update(atomic=[User.login_count.if_not_exists(0)])
    # login_count: 0

    # Now increment it
    await user.update(atomic=[User.login_count.add(1)])
    # login_count: 1

    # if_not_exists won't overwrite existing value
    await user.update(atomic=[User.login_count.if_not_exists(999)])
    # login_count: still 1

    # add() already initializes from zero, so no if_not_exists is needed
    await user.update(atomic=[User.score.add(10)])
    # score: 10


asyncio.run(main())

Multiple operations

Combine several atomic operations in one request:

multiple_ops.py

"""Multiple atomic operations in one request."""

import asyncio
from datetime import datetime

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


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

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    login_count = NumberAttribute()
    last_login = StringAttribute()
    badges = ListAttribute()
    temp_token = StringAttribute()


async def main():
    user = User(
        pk="USER#123",
        sk="PROFILE",
        login_count=0,
        badges=[],
        temp_token="abc123",
    )
    await user.save()

    # Multiple operations in one request
    await user.update(
        atomic=[
            User.login_count.add(1),
            User.last_login.set(datetime.now().isoformat()),
            User.badges.append(["first_login"]),
            User.temp_token.remove(),
        ]
    )

    # Result:
    # login_count: 1
    # last_login: "2024-01-15T10:30:00"
    # badges: ["first_login"]
    # temp_token: None (removed)


asyncio.run(main())

All operations happen atomically. Either all succeed or none do.

Operations reference

Method	Description	Example
`set(value)`	Set attribute to value	`User.name.set("Jane")`
`add(n)`	Add to number, starting from zero if missing (use negative to subtract)	`User.count.add(1)`
`remove()`	Delete the attribute	`User.temp.remove()`
`append(items)`	Add items to end of list	`User.tags.append(["a", "b"])`
`prepend(items)`	Add items to start of list	`User.tags.prepend(["a"])`
`if_not_exists(value)`	Set only if attribute is missing	`User.count.if_not_exists(0)`

Error handling

When a condition fails, you get ConditionalCheckFailedException:

from pydynox.exceptions import ConditionalCheckFailedException

try:
    account.update(
        atomic=[Account.balance.add(-100)],
        condition=Account.balance >= 100,
    )
except ConditionalCheckFailedException:
    print("Insufficient balance")

When to use atomic updates

Use atomic updates when:

Multiple requests might update the same item
You need counters (views, likes, inventory)
You want to avoid read-modify-write patterns
You need guaranteed consistency

Use regular update() with kwargs when:

You're the only writer
You need to set values based on other fields
You're doing a simple field update

Testing your code

Test atomic updates without DynamoDB using the built-in memory backend:

testing_atomic.py

"""Testing atomic updates with pydynox_memory_backend."""

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


class Counter(Model):
    model_config = ModelConfig(table="counters")
    pk = StringAttribute(partition_key=True)
    count = NumberAttribute(default=0)
    tags = ListAttribute(default=list)


def test_atomic_increment(pydynox_memory_backend):
    """Test atomic counter increment."""
    counter = Counter(pk="views", count=0)
    counter.save()

    # Increment atomically
    counter.update(atomic=[Counter.count.add(1)])
    counter.update(atomic=[Counter.count.add(1)])
    counter.update(atomic=[Counter.count.add(1)])

    found = Counter.get(pk="views")
    assert found.count == 3


def test_atomic_decrement(pydynox_memory_backend):
    """Test atomic counter decrement."""
    counter = Counter(pk="stock", count=100)
    counter.save()

    # Decrement atomically
    counter.update(atomic=[Counter.count.add(-10)])

    found = Counter.get(pk="stock")
    assert found.count == 90


def test_atomic_append(pydynox_memory_backend):
    """Test atomic list append."""
    counter = Counter(pk="item", tags=["initial"])
    counter.save()

    # Append atomically
    counter.update(atomic=[Counter.tags.append(["new_tag"])])

    found = Counter.get(pk="item")
    assert "initial" in found.tags
    assert "new_tag" in found.tags


def test_atomic_set(pydynox_memory_backend):
    """Test atomic set operation."""
    counter = Counter(pk="item", count=10)
    counter.save()

    # Set atomically
    counter.update(atomic=[Counter.count.set(99)])

    found = Counter.get(pk="item")
    assert found.count == 99

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

Next steps

Observability - Track metrics on every operation
Conditions - Add conditions to atomic updates
Transactions - Combine atomic updates in transactions