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.
|
🧬 Pydantic v2 Models — full type-checking & validation |
🔄 Automatic Schema Sync — |
| 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 |
❌ | ❌ |
| Vector Search (ANN) | ✅ Vector() + order_by_ann() |
❌ | ❌ |
| Python Version | 3.10+ | 3.8+ | 3.6+ |
pip install coodieChoose 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 acsylla1. 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="alice@example.com")
user.save()
# Query
print(User.find(name="Alice").allow_filtering().all())💡 Async? Just swap
coodie.syncforcoodie.aioand addawait— that's it!
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 |
Vector(dimensions=N) |
Vector column — maps list[float] to vector<float, N> |
VectorIndex(similarity_function="COSINE") |
SAI vector index for ANN queries |
Vector Search (ANN)
coodie supports ScyllaDB's vector<float, N> column type and ANN similarity search via VectorIndex.
class ProductEmbedding(Document):
product_id: Annotated[UUID, PrimaryKey()] = Field(default_factory=uuid4)
embedding: Annotated[
list[float],
Vector(dimensions=384),
VectorIndex(similarity_function="COSINE"),
]
...
results = await ProductEmbedding.find().order_by_ann("embedding", query_vector).limit(10).all()📖 Full guide: Vector Search (ANN)
| Resource | Link |
|---|---|
| 📖 Full Documentation | scylladb.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 |
Thanks goes to these wonderful people (emoji key):
 Israel Fruchter 💻 🤔 📖 |
This project follows the all-contributors specification. Contributions of any kind welcome!
This package was created with Cookiecutter and the browniebroke/cookiecutter-pypackage project template.