Python code to do a insert in scylladb using python having custom frozen field

Question

My db was created with below configuration

CREATE KEYSPACE chat WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1}; use chat;

CREATE TYPE IF NOT EXISTS conversation ( id TEXT, message TEXT, audio TEXT, metadata MAP<TEXT, TEXT>, timestamp TIMESTAMP );

CREATE TABLE IF NOT EXISTS chat ( user_id TEXT, interaction_id TEXT, created_timestamp TIMESTAMP, prompt frozen, response frozen, interaction_details MAP<TEXT, TEXT>, regeneration_details LIST<frozen>, PRIMARY KEY (user_id, interaction_id, created_timestamp) );

I have written below code to a simple insert in scylladb using python

import uuid
import traceback
from datetime import datetime, timedelta, timezone
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from cassandra import ConsistencyLevel
from config import db_config as db

def connect_to_cassandra(local_port, cassandra_username, cassandra_password, keyspace):
    auth_provider = PlainTextAuthProvider(username=cassandra_username, password=cassandra_password)
    cluster = Cluster(['127.0.0.1'], port=local_port, auth_provider=auth_provider)
    session = cluster.connect(keyspace=keyspace)
    session.default_consistency_level = ConsistencyLevel.QUORUM
    return session, cluster


local_port = db.LOCAL_PORT
remote_cassandra_host = db.REMOTE_CASSANDRA_HOST
remote_cassandra_port = int(db.REMOTE_CASSANDRA_PORT)
cassandra_username = db.CASSANDRA_USERNAME
cassandra_password = db.CASSANDRA_PASSWORD
cassandra_keyspace = db.CASSANDRA_KEYSPACE


user_id = '735f0735-8851-45b4-b0a3-dae7061b024a'
interaction_id = str(uuid.uuid4())
created_timestamp = datetime.now() - timedelta(days=365)
interaction_details = {"title": "Hi"}

prompt_data = {
    "id": str(uuid.uuid4()),
    "message": "Hello",
    "audio": None,
    "metadata": {'type': 'text'},
    "timestamp": created_timestamp
    }

response_data = {
    "id": str(uuid.uuid4()),
    "message": "Hi",
    "audio": None,
    "metadata": {"type": "text"},
    "timestamp": created_timestamp
    }
response_data = None
regeneration_details_data = None

insert_query = "INSERT INTO chat (user_id, interaction_id, created_timestamp, interaction_details, prompt, regeneration_details, response) VALUES (?, ?, ?, ?, ?, ?, ?)"

try:
    cassandra_session = connect_to_cassandra(local_port, cassandra_username, cassandra_password, cassandra_keyspace)
except Exception:
    print(traceback.format_exc())

insert_ps=cassandra_session.prepare(query=insert_query)
cassandra_session.execute(query=insert_ps, parameters=[user_id, interaction_id, created_timestamp, interaction_details, prompt_data,regeneration_details_data,response_data])

I am getting error

cassandra_session.execute( File "cassandra/cluster.py", line 2677, in cassandra.cluster.Session.execute File "cassandra/cluster.py", line 2720, in cassandra.cluster.Session.execute_async File "cassandra/cluster.py", line 2979, in cassandra.cluster.Session._create_response_future File "cassandra/query.py", line 905, in cassandra.query.bind_params TypeError: not all arguments converted during string formatting

If i set response_data to None. Then query get executed fine. But if I try to pass response_data, then I am getting error. What am I doing wrong?

Mentioned in above details

Answer 1

The Python driver documentation in https://docs.datastax.com/en/developer/python-driver/3.28/user_defined_types/ explains how to pass user-defined types as bind variables. The documentation suggests a few options, such as creating a special class, or "registering" the user-defined type, but you can also do it without a class or registrations, using a named tuple. I don't think it just works naively with a dict, e.g.,

import collections
# user_type("a", 1, "b", 2) creates a named tuple with component names "a", "b"
# and values 1, 2. The return of this function can be used to bind to a UDT.
# The number of arguments is assumed to be even.
def user_type(*args):
    return collections.namedtuple('user_type', args[::2])(*args[1::2])
...
execute(stmt, [user_type('a', 0, 'b', 2)])