pydynox 0.1.0

A fast DynamoDB ORM for Python with a Rust core

pydynox 🐍⚙️

A fast DynamoDB ORM for Python with a Rust core.

⚠️ Alpha Software: This project is in early development. The API may change before v1.0. Use in production at your own risk. Contributions are welcome!

Why "pydynox"?

Py(thon) + Dyn(amoDB) + Ox(ide/Rust)

Features

• Simple class-based API like PynamoDB
• Fast serialization with Rust
• Batch operations with auto-splitting
• Transactions
• Global Secondary Indexes
• Async support
• Pydantic integration

Performance

pydynox is faster than PynamoDB and boto3 in all operations. Benchmarks run against moto (local DynamoDB mock):

Operation	pydynox	PynamoDB	boto3	vs PynamoDB	vs boto3
batch_get (100 items)	2.55ms	4.24ms	3.92ms	1.66x faster	1.54x faster
batch_write (100 items)	7.74ms	9.53ms	9.68ms	1.23x faster	1.25x faster
get_item (10x)	13.83ms	16.57ms	16.86ms	1.20x faster	1.22x faster
put_item (10x)	14.13ms	16.78ms	16.80ms	1.19x faster	1.19x faster
query (100 items)	12.92ms	14.88ms	14.79ms	1.15x faster	1.15x faster
update_item (10x)	19.04ms	38.15ms	21.81ms	2.00x faster	1.15x faster
delete_item (10x)	26.96ms	33.07ms	32.93ms	1.23x faster	1.22x faster

Highlights:

• batch_get: 66% faster than PynamoDB
• update_item: 2x faster than PynamoDB

Run benchmarks yourself:

uv run maturin develop --release
uv run pytest benchmark/benchmark.py -v --benchmark-only

Installation

pip install pydynox

For Pydantic support:

pip install pydynox[pydantic]

Quick Start

Define a Model

from pydynox import Model, String, Number, Boolean, List

class User(Model):
    class Meta:
        table = "users"
    
    pk = String(hash_key=True)
    sk = String(range_key=True)
    name = String()
    email = String()
    age = Number(default=0)
    active = Boolean(default=True)
    tags = List(String)

CRUD Operations

# Create
user = User(pk="USER#123", sk="PROFILE", name="John", email="[e-mail]")
user.save()

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

# Update
user.name = "John Doe"
user.save()

# Delete
user.delete()

Query

from pydynox import Condition

# Simple query
users = User.query(pk="USER#123")

# With filters
users = User.query(pk="USER#123") \
    .where(Condition.begins_with("sk", "ORDER#")) \
    .where(Condition.gt("age", 18)) \
    .exec()

# Iterate (auto pagination)
for user in users:
    print(user.name)

Conditions

from pydynox import Condition

# Save with condition
user.save(condition=Condition.not_exists("pk"))

# Delete with condition
user.delete(condition=Condition.eq("version", 5))

# Combine conditions
user.save(
    condition=Condition.not_exists("pk") | Condition.eq("version", 1)
)

Available conditions:

• Condition.eq(field, value) - equals
• Condition.ne(field, value) - not equals
• Condition.gt(field, value) - greater than
• Condition.gte(field, value) - greater than or equal
• Condition.lt(field, value) - less than
• Condition.lte(field, value) - less than or equal
• Condition.exists(field) - attribute exists
• Condition.not_exists(field) - attribute does not exist
• Condition.begins_with(field, prefix) - string starts with
• Condition.contains(field, value) - string or list contains
• Condition.between(field, low, high) - value in range

Atomic Updates

from pydynox import Action

# Simple set
user.update(name="New Name", email="[e-mail]")

# Increment a number
user.update(Action.increment("age", 1))

# Append to list
user.update(Action.append("tags", ["verified"]))

# Remove field
user.update(Action.remove("temp_field"))

# Combine with condition
user.update(
    Action.increment("age", 1),
    condition=Condition.eq("status", "active")
)

Batch Operations

# Batch write
with User.batch_write() as batch:
    batch.save(user1)
    batch.save(user2)
    batch.delete(user3)

# Batch get
users = User.batch_get([
    ("USER#1", "PROFILE"),
    ("USER#2", "PROFILE"),
])

Global Secondary Index

from pydynox import GlobalIndex

class User(Model):
    class Meta:
        table = "users"
    
    pk = String(hash_key=True)
    sk = String(range_key=True)
    email = String()
    
    email_index = GlobalIndex(hash_key="email")

# Query on index
users = User.email_index.query(email="[e-mail]")

Transactions

with User.transaction() as tx:
    tx.save(user1)
    tx.delete(user2)
    tx.update(user3, Action.increment("age", 1))

Async Support

# All methods work with await
user = await User.get(pk="USER#123", sk="PROFILE")
await user.save()

async for user in User.query(pk="USER#123"):
    print(user.name)

Pydantic Integration

from pydynox import dynamodb_model
from pydantic import BaseModel, EmailStr

@dynamodb_model(table="users", hash_key="pk", range_key="sk")
class User(BaseModel):
    pk: str
    sk: str
    name: str
    email: EmailStr
    age: int = 0

# All pydynox methods available
user = User(pk="USER#123", sk="PROFILE", name="John", email="[e-mail]")
user.save()

Table Management

# Create table
User.create_table()

# Create with custom capacity
User.create_table(read_capacity=10, write_capacity=5)

# Create with on-demand billing
User.create_table(billing_mode="PAY_PER_REQUEST")

# Check if table exists
if not User.table_exists():
    User.create_table()

# Delete table
User.delete_table()

Documentation

Full documentation: https://pydynox.readthedocs.io

License

Apache 2.0 License