pydynox 0.14.0

A fast DynamoDB ORM for Python with a Rust core

pydynox 🐍⚙️

A fast DynamoDB ORM for Python with a Rust core.

Pre-release: The core features are working and tested. We're adding features, polishing the API, receiving ideas, and testing performance and edge cases before v1.0. Feel free to try it out and share feedback!

Why "pydynox"?

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

GenAI Contributions 🤖

I believe GenAI is transforming how we build software. It's a powerful tool that accelerates development when used by developers who understand what they're doing.

To support both humans and AI agents, I created:

• .ai/ folder - Guidelines for agentic IDEs (Cursor, Windsurf, Kiro, etc.)
• ADR/ folder - Architecture Decision Records for humans to understand the "why" behind decisions

If you're contributing with AI help:

• Understand what the AI generated before submitting
• Make sure the code follows the project patterns
• Test your changes

I reserve the right to reject low-quality PRs where project patterns are not followed and it's clear that GenAI was driving instead of the developer.

Features

• Simple class-based API like PynamoDB
• Fast serialization with Rust
• Batch operations with auto-splitting
• Transactions
• Global Secondary Indexes
• Async support
• Pydantic integration
• TTL (auto-expiring items)
• Lifecycle hooks
• Auto-generate IDs and timestamps
• Optimistic locking
• Rate limiting
• Field encryption (KMS)
• Compression (zstd, lz4, gzip)
• S3 attribute for large files
• PartiQL support
• Observability (logging, metrics)

Installation

pip install pydynox

For Pydantic support:

pip install pydynox[pydantic]

Quick Start

Define a Model

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

class User(Model):
    model_config = ModelConfig(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, ModelConfig

class User(Model):
    model_config = ModelConfig(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()

S3 Attribute (Large Files)

DynamoDB has a 400KB item limit. S3Attribute stores files in S3 and keeps metadata in DynamoDB. Upload on save, download on demand, delete when the item is deleted.

from pydynox.attributes import S3Attribute
from pydynox._internal._s3 import S3File

class Document(Model):
    model_config = ModelConfig(table="documents")
    
    pk = StringAttribute(hash_key=True)
    content = S3Attribute(bucket="my-bucket", prefix="docs/")

# Upload
doc = Document(pk="DOC#1")
doc.content = S3File(b"...", name="report.pdf", content_type="application/pdf")
doc.save()

# Download
doc = Document.get(pk="DOC#1")
data = doc.content.get_bytes()           # Load to memory
doc.content.save_to("/path/to/file.pdf") # Stream to file
url = doc.content.presigned_url(3600)    # Share via URL

# Metadata (no S3 call)
print(doc.content.size)
print(doc.content.content_type)

# Delete - removes from both DynamoDB and S3
doc.delete()

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://leandrodamascena.github.io/pydynox

License

MIT License

Inspirations

This project was inspired by:

• PynamoDB - The ORM-style API and model design
• Pydantic - Data validation patterns and integration approach
• dynarust - Rust DynamoDB client patterns
• dyntastic - Pydantic + DynamoDB integration ideas

Building from Source

Requirements

• Python 3.11+
• Rust 1.70+
• maturin

Setup

# Clone the repo
git clone https://github.com/leandrodamascena/pydynox.git
cd pydynox

# Install maturin
pip install maturin

# Build and install locally
maturin develop

# Or with uv
uv run maturin develop

Running Tests

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest