Use pydantic models to create basic CQL queries.
Project description
Pyssandra
Cassandra object mapper using 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})
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
pyssandra-0.12.4.tar.gz
(7.3 kB
view hashes)
Built Distribution
Close
Hashes for pyssandra-0.12.4-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | fbd277b38b78506d9bd200938c36d5e1b30ccdbf9b90c549a0556e66e19330db |
|
MD5 | b0de76535272e6bf387ab2351a996a1c |
|
BLAKE2b-256 | 9928bb54e31c68bece42005bd48b1c5945119beafa599a43648b266c3b26cd33 |