Conditions

Conditions let you add rules to save() and delete() operations. The operation only happens if the condition is true. If not, DynamoDB raises an error.

This is useful for:

Preventing overwrites (only save if item doesn't exist)
Optimistic locking (only update if version matches)
Business rules (only withdraw if balance is sufficient)

Key features

Use Python operators (==, >, <) on model attributes
Combine with & (AND), | (OR), ~ (NOT)
Functions like exists(), begins_with(), between()
Build conditions dynamically from user input

Getting started

Prevent overwriting existing items

The most common use case. Only save if the item doesn't exist yet:

prevent_overwrite.py

"""Prevent overwriting existing items."""

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)
    email = StringAttribute()
    name = StringAttribute()
    age = NumberAttribute()


async def main():
    # Only save if the item doesn't exist yet
    user = User(pk="USER#NEW", sk="PROFILE", email="john@example.com", name="John", age=30)
    await user.save(condition=User.pk.not_exists())

    # If USER#NEW already exists, this raises ConditionalCheckFailedException


asyncio.run(main())

Without this condition, save() would silently overwrite any existing item with the same key. With the condition, you get a ConditionalCheckFailedException if the item already exists.

Safe delete

Only delete if certain conditions are met:

safe_delete.py

"""Safe delete - only delete if conditions are met."""

import asyncio

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


class Order(Model):
    model_config = ModelConfig(table="orders")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    status = StringAttribute()
    total = NumberAttribute()


async def main():
    # Create an order first
    order = Order(pk="ORDER#123", sk="DETAILS", status="draft", total=100)
    await order.save()

    # Only delete if order is in "draft" status
    await order.delete(condition=Order.status == "draft")

    # Can't delete orders that are already processed


asyncio.run(main())

This prevents accidental deletion of orders that are already being processed.

Advanced

Optimistic locking

When multiple processes might update the same item, use a version field to prevent lost updates:

optimistic_lock.py

"""Optimistic locking - only update if version matches."""

import asyncio

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


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

    pk = StringAttribute(partition_key=True)
    name = StringAttribute()
    price = NumberAttribute()
    version = NumberAttribute()


async def main():
    # Create product first
    product = Product(pk="PROD#123", name="Widget", price=19.99, version=1)
    await product.save()

    # Get current product
    product = await Product.get(pk="PROD#123")
    current_version = product.version

    # Update with version check
    product.price = 29.99
    product.version = current_version + 1
    await product.save(condition=Product.version == current_version)

    # If someone else updated the product, version won't match
    # and ConditionalCheckFailedException is raised


asyncio.run(main())

How it works:

Read the item and note the current version
Make your changes and increment the version
Save with a condition that the version still matches
If someone else updated it first, the condition fails

This is safer than "last write wins" because you know when conflicts happen.

Complex business rules

Combine multiple conditions for complex validation:

complex_rules.py

"""Complex business rules with combined conditions."""

import asyncio

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


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

    pk = StringAttribute(partition_key=True)
    balance = NumberAttribute()
    status = StringAttribute()
    verified = BooleanAttribute()


async def main():
    # Create account first
    account = Account(pk="ACC#123", balance=500, status="active", verified=True)
    await account.save()

    # Only allow withdrawal if:
    # - Account is active AND verified
    # - Balance is sufficient
    withdrawal = 100

    condition = (
        (Account.status == "active")
        & (Account.verified == True)  # noqa: E712
        & (Account.balance >= withdrawal)
    )

    account.balance = account.balance - withdrawal
    await account.save(condition=condition)


asyncio.run(main())

The withdrawal only happens if all three conditions are true. If any fails, the whole operation is rejected.

Dynamic filters

Build conditions at runtime based on user input:

dynamic_filters.py

"""Build conditions dynamically from user input."""

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


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

    pk = StringAttribute(partition_key=True)
    category = StringAttribute()
    price = NumberAttribute()
    brand = StringAttribute()


def build_product_filter(
    category: str | None = None,
    max_price: float | None = None,
    brand: str | None = None,
):
    """Build filter from optional parameters."""
    conditions = []

    if category:
        conditions.append(Product.category == category)
    if max_price:
        conditions.append(Product.price <= max_price)
    if brand:
        conditions.append(Product.brand == brand)

    if len(conditions) == 0:
        return None
    if len(conditions) == 1:
        return conditions[0]

    return And(*conditions)


# User searches for electronics under $500
filter_cond = build_product_filter(category="electronics", max_price=500)

Use And() and Or() from pydynox.conditions when you have a list of conditions to combine.

Operators

Comparison operators

Operator	Example	Description
`==`	`User.status == "active"`	Equal
`!=`	`User.status != "deleted"`	Not equal
`>`	`User.age > 18`	Greater than
`>=`	`User.age >= 21`	Greater or equal
`<`	`User.age < 65`	Less than
`<=`	`User.age <= 30`	Less or equal

Combining operators

Operator	Example	Description
`&`	`(a > 1) & (b < 2)`	Both must be true
`\\|`	`(a == 1) \\| (a == 2)`	Either can be true
`~`	`~a.exists()`	Negates the condition

Warning

Always use parentheses when combining conditions. Python's operator precedence may not work as expected.

Function conditions

Function	Example	Description
`exists()`	`User.email.exists()`	Attribute exists
`does_not_exist()`	`User.pk.not_exists()`	Attribute doesn't exist
`begins_with()`	`User.sk.begins_with("ORDER#")`	String starts with prefix
`contains()`	`User.tags.contains("vip")`	List contains value
`between()`	`User.age.between(18, 65)`	Value in range (inclusive)
`is_in()`	`User.status.is_in("a", "b")`	Value in list

Nested attributes

Access nested map keys and list indexes:

# Map access
User.address["city"] == "NYC"

# List access
User.tags[0] == "premium"

# Deep nesting
User.metadata["preferences"]["theme"] == "dark"

Error handling

When a condition fails, DynamoDB raises ConditionalCheckFailedException:

from pydynox.exceptions import ConditionalCheckFailedException

try:
    await user.save(condition=User.pk.not_exists())
except ConditionalCheckFailedException:
    print("User already exists")

Get the existing item on failure

Use return_values_on_condition_check_failure=True to get the existing item without an extra GET call:

try:
    await client.put_item(
        "users",
        {"pk": "USER#123", "name": "Bob"},
        condition_expression="attribute_not_exists(pk)",
        return_values_on_condition_check_failure=True,
    )
except ConditionalCheckFailedException as e:
    print(f"User already exists: {e.item}")

This works with put_item, update_item, and delete_item.

Type hints

Use Condition for type hints:

from pydynox import Condition

def apply_filter(cond: Condition) -> None:
    ...

Testing your code

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

testing_conditions.py

"""Testing conditions with pydynox_memory_backend."""

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


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    name = StringAttribute()
    status = StringAttribute(default="active")
    version = NumberAttribute(default=1)


def test_prevent_overwrite(pydynox_memory_backend):
    """Test condition to prevent overwriting existing items."""
    user = User(pk="USER#1", name="John")
    user.save(condition=User.pk.not_exists())

    # Second save with same key should fail
    user2 = User(pk="USER#1", name="Jane")
    with pytest.raises(ConditionalCheckFailedException):
        user2.save(condition=User.pk.not_exists())


def test_conditional_delete(pydynox_memory_backend):
    """Test conditional delete."""
    user = User(pk="USER#1", name="John", status="inactive")
    user.save()

    # Can only delete inactive users
    user.delete(condition=User.status == "inactive")

    assert User.get(pk="USER#1") is None


def test_conditional_delete_fails(pydynox_memory_backend):
    """Test that conditional delete fails when condition not met."""
    user = User(pk="USER#1", name="John", status="active")
    user.save()

    # Cannot delete active users
    with pytest.raises(ConditionalCheckFailedException):
        user.delete(condition=User.status == "inactive")

    # User still exists
    assert User.get(pk="USER#1") is not None


def test_optimistic_locking(pydynox_memory_backend):
    """Test optimistic locking with version field."""
    user = User(pk="USER#1", name="John", version=1)
    user.save()

    # Simulate concurrent update
    user1 = User.get(pk="USER#1")
    user2 = User.get(pk="USER#1")

    # First update succeeds
    user1.name = "Jane"
    user1.version = 2
    user1.save(condition=User.version == 1)

    # Second update fails (version already changed)
    user2.name = "Bob"
    user2.version = 2
    with pytest.raises(ConditionalCheckFailedException):
        user2.save(condition=User.version == 1)

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

Next steps

Query - Query items with conditions
Atomic updates - Increment, append, and other atomic operations
Transactions - All-or-nothing operations