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."""

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


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

    pk = StringAttribute(hash_key=True)  # page_url
    sk = StringAttribute(range_key=True)  # date
    views = NumberAttribute()


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

# Each request increments atomically
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

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

Safe balance transfer

Combine atomic updates with conditions to prevent overdrafts:

balance_transfer.py

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

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


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

    pk = StringAttribute(hash_key=True)  # account_id
    sk = StringAttribute(range_key=True)  # "BALANCE"
    balance = NumberAttribute()


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


# Usage
account = Account(pk="ACC#123", sk="BALANCE", balance=100)
account.save()

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

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

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."""

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


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

    pk = StringAttribute(hash_key=True)  # product_id
    sk = StringAttribute(range_key=True)  # "STOCK"
    stock = NumberAttribute()
    reserved = NumberAttribute()


class OutOfStock(Exception):
    pass


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


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


# Usage
product = Product(pk="SKU#ABC123", sk="STOCK", stock=10, reserved=0)
product.save()

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

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

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

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."""

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


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

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


class RateLimitExceeded(Exception):
    pass


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.get(pk=user_id, sk=date)

    if usage is None:
        # First request of the day
        usage = ApiUsage(pk=user_id, sk=date, requests=1)
        usage.save()
        return 1

    try:
        usage.update(
            atomic=[ApiUsage.requests.add(1)],
            condition=ApiUsage.requests < daily_limit,
        )
        # Fetch updated count
        updated = ApiUsage.get(pk=user_id, sk=date)
        return updated.requests
    except ConditionCheckFailedError:
        raise RateLimitExceeded(f"User {user_id} exceeded {daily_limit} requests/day")


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

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."""

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


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

    pk = StringAttribute(hash_key=True)  # user_id
    sk = StringAttribute(range_key=True)  # "CART"
    items = ListAttribute()
    total = NumberAttribute()


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


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


# Usage
cart = Cart(pk="USER#123", sk="CART", items=[], total=0)
cart.save()

# Add items
add_to_cart(cart, {"sku": "SHIRT-M", "qty": 1}, 29.99)
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
apply_discount(cart, 10.00)
# total: 69.98

List operations

Add items to lists without reading the whole list:

user_tags.py

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

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


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

    pk = StringAttribute(hash_key=True)
    sk = StringAttribute(range_key=True)
    tags = ListAttribute()


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

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

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

Default values

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

if_not_exists.py

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

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


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

    pk = StringAttribute(hash_key=True)
    sk = StringAttribute(range_key=True)
    login_count = NumberAttribute(null=True)
    score = NumberAttribute(null=True)


# User without login_count
user = User(pk="USER#123", sk="PROFILE")
user.save()

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

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

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

# Combine with add for "increment or initialize"
user.update(
    atomic=[
        User.score.if_not_exists(0),  # Initialize if missing
    ]
)
user.update(atomic=[User.score.add(10)])  # Then increment
# score: 10

Multiple operations

Combine several atomic operations in one request:

multiple_ops.py

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

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(hash_key=True)
    sk = StringAttribute(range_key=True)
    login_count = NumberAttribute()
    last_login = StringAttribute(null=True)
    badges = ListAttribute()
    temp_token = StringAttribute(null=True)


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

# Multiple operations in one request
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)

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 (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 ConditionCheckFailedError:

from pydynox.exceptions import ConditionCheckFailedError

try:
    account.update(
        atomic=[Account.balance.add(-100)],
        condition=Account.balance >= 100,
    )
except ConditionCheckFailedError:
    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