pydynox 0.2.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!

📋 Check the open issues to see what's planned and what's being worked on.

Why "pydynox"?

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

GenAI Contributions 🤖

This project welcomes GenAI-assisted contributions! Parts of the code and docstrings were written with AI help.

If you're using AI tools to contribute:

• Follow the project's coding style and patterns
• Make sure the code makes sense and is tested
• Don't submit random or low-quality generated code
• Review and understand what the AI generated before submitting

Features

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

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