pyssandra 0.10.5

Use pydantic models to create basic CQL queries.

Pyssandra

Build simple CQL queries from Pydantic models.

Example

import uuid

from cassandra.auth import PlainTextAuthProvider
from cassandra.cluster import Cluster
from pydantic import BaseModel, Field

from pyssandra import Pyssandra

cloud_config = {"secure_connect_bundle": "/path/to/secure-connect-dbname.zip"}
auth_provider = PlainTextAuthProvider(username="user", password="pass")
cluster = Cluster(cloud=cloud_config, auth_provider=auth_provider)
session = cluster.connect()

keyspace = "test"
db = Pyssandra(session, keyspace)


@db.table(partition_keys=["id"], index=["first"])
class User(BaseModel):
    """Test user model."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    first: str
    last: str


# Create DB Table.
db[User].sync()

# CRUD Methods.
user = User(first="Test", last="User")
await db[User].insert(user)
await db[User].find_one({"id": user.id})
page1 = await db[User].find_many(where={"first": "Test"}, page_size=10)
page2 = await db[User].find_many(where={"first": "Test"}, page_size=10, paging_state=page1.paging_state)
await db[User].update(user)
await db[User].delete({"id": user.id})