coodie

The modern Pydantic-based ODM for Cassandra & ScyllaDB

Cassandra + Beanie (hoodie) = coodie 🧥

📖 Documentation • 🚀 Quick Start • 🤝 Contributing • 📋 Changelog

---

Define your data models as Python classes, and coodie handles schema synchronization, serialization, and query building — with both sync and async APIs.

✨ Feature Highlights

🧬 Pydantic v2 Models — full type-checking & validation ⚡ Sync & Async — coodie.sync for blocking, coodie.aio for asyncio 🔗 Chainable QuerySet — .filter() · .limit() · .order_by()

🔄 Automatic Schema Sync — sync_table() creates & evolves tables 🏗️ Batch & LWT — BatchQuery + if_not_exists() support 🎯 Multi-Driver — scylla-driver · cassandra-driver · acsylla

🔍 How Does coodie Compare?

Feature	coodie	beanie	cqlengine
Database	Cassandra / ScyllaDB	MongoDB	Cassandra
Schema Definition	Pydantic v2 BaseModel	Pydantic v2 BaseModel	Custom columns.* classes
Type Hints	✅ Native Annotated[]	✅ Native Pydantic	❌ No type hints
Async Support	✅ First-class	✅ First-class	❌ Sync only
Sync Support	✅ coodie.sync	❌ Async only	✅ Sync only
Query API	Chainable QuerySet	Chainable FindMany	Chainable QuerySet
Schema Migration	✅ sync_table()	❌ Manual	✅ sync_table()
LWT (Compare-and-Set)	✅ if_not_exists()	N/A	✅ iff()
Batch Operations	✅ BatchQuery	❌	✅ BatchQuery
Counter Columns	✅ Counter()	❌	✅ columns.Counter
User-Defined Types	✅ UserType	❌	✅ UserType
TTL Support	✅ Per-save TTL	❌	✅ Per-save TTL
Pagination	✅ Token-based PagedResult	✅ Cursor-based	❌ Manual
Multiple Drivers	✅ 3 drivers	motor only	cassandra-driver only
Polymorphic Models	✅ Discriminator	❌	❌
Python Version	3.10+	3.8+	3.6+

📦 Installation

pip install coodie

Choose a driver extra for your cluster:

pip install "coodie[scylla]"      # ScyllaDB / Cassandra (recommended)
pip install "coodie[cassandra]"   # Cassandra via cassandra-driver
pip install "coodie[acsylla]"     # Async-native via acsylla

🚀 Quick Start

1. Start a local ScyllaDB (or use an existing cluster):

docker run --name scylla -d -p 9042:9042 scylladb/scylla --smp 1

# Wait for it to be ready (~30s), then create a keyspace
docker exec -it scylla cqlsh -e \
  "CREATE KEYSPACE IF NOT EXISTS my_ks
   WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};"

2. Install coodie:

pip install "coodie[scylla]"

3. Write your first script:

from coodie.sync import Document, init_coodie
from coodie.fields import PrimaryKey
from pydantic import Field
from typing import Annotated
from uuid import UUID, uuid4

# Connect
init_coodie(hosts=["127.0.0.1"], keyspace="my_ks")

# Define a model
class User(Document):
    id: Annotated[UUID, PrimaryKey()] = Field(default_factory=uuid4)
    name: str
    email: str

    class Settings:
        name = "users"

# Sync schema & insert
User.sync_table()
user = User(name="Alice", email="[email protected]")
user.save()

# Query
print(User.find(name="Alice").allow_filtering().all())

💡 Async? Just swap coodie.sync for coodie.aio and add await — that's it!

📖 Usage

Define a Document

from typing import Annotated
from uuid import UUID, uuid4
from pydantic import Field
from coodie import Document, PrimaryKey, ClusteringKey, Indexed

class Product(Document):
    id: Annotated[UUID, PrimaryKey()] = Field(default_factory=uuid4)
    category: Annotated[str, ClusteringKey()] = "general"
    name: str
    brand: Annotated[str, Indexed()] = "Unknown"
    price: float = 0.0

    class Settings:
        name = "products"      # table name (defaults to snake_case class name)
        keyspace = "my_ks"

Async API — coodie / coodie.aio

import asyncio
from coodie import init_coodie
# Product defined above — same field definitions, using coodie.aio.Document

async def main():
    await init_coodie(hosts=["127.0.0.1"], keyspace="my_ks")
    await Product.sync_table()

    p = Product(name="Gadget", brand="Acme", price=9.99)
    await p.save()

    results = await Product.find(brand="Acme").limit(10).all()
    for product in results:
        print(product.name, product.price)

    gadget = await Product.get(id=p.id)
    await gadget.delete()

asyncio.run(main())

Sync API — coodie.sync

from coodie.sync import Document, init_coodie

class Product(Document):
    ...  # same field definitions

init_coodie(hosts=["127.0.0.1"], keyspace="my_ks")
Product.sync_table()

p = Product(name="Widget", price=4.99)
p.save()

results = Product.find(brand="Acme").allow_filtering().all()
one = Product.find_one(name="Widget")

QuerySet Chaining

# Filter, sort, and limit
products = (
    await Product.find()
    .filter(brand="Acme")
    .order_by("price")
    .limit(20)
    .all()
)

# Count
n = await Product.find(brand="Acme").allow_filtering().count()

# Async iteration
async for p in Product.find(brand="Acme"):
    print(p)

# Delete matching rows
await Product.find(brand="Discontinued").allow_filtering().delete()

Field Annotations Reference

Annotation	Purpose
PrimaryKey(partition_key_index=0)	Partition key column (composite keys via index)
ClusteringKey(order="ASC", clustering_key_index=0)	Clustering column
Indexed(index_name=None)	Secondary index
Counter()	Counter column
Discriminator()	Polymorphic model discriminator

📚 Learn More

Resource	Link
📖 Full Documentation	fruch.github.io/coodie
🚀 Quick Start Guide	Installation & Quickstart
📊 Benchmark History	Performance Trends
🔄 Migrating from cqlengine	Migration Guide
🤝 Contributing	CONTRIBUTING.md
📋 Changelog	CHANGELOG.md
🐛 Bug Reports	GitHub Issues

Contributors ✨

Thanks goes to these wonderful people (emoji key):

Israel Fruchter 💻 🤔 📖

This project follows the all-contributors specification. Contributions of any kind welcome!

Credits

This package was created with Cookiecutter and the browniebroke/cookiecutter-pypackage project template.