使用具有自定义冻结字段的 python 在 scyladb 中执行插入的 Python 代码

Question

我的数据库是使用以下配置创建的

使用 REPLICATION 创建 KEYSPACE 聊天 = {'class': 'SimpleStrategy', 'replication_factor': 1}; 使用聊天；

如果不存在则创建类型对话（ ID 文本， 消息文本， 音频文本， 元数据地图， 时间戳 时间戳 );

如果不存在则创建表聊天（ 用户 ID 文本， 交互 ID 文本， 创建时间戳时间戳， 迅速冻结， 反应冻结， 互动详情地图， 再生_详细信息列表， 主键（用户 ID、交互 ID、创建时间戳） );

我已经使用 python 将下面的代码写入了 scyladb 中的简单插入

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])

我收到错误

cassandra_session.execute( 文件“cassandra/cluster.py”，第 2677 行，位于 cassandra.cluster.Session.execute 文件“cassandra/cluster.py”，第 2720 行，位于 cassandra.cluster.Session.execute_async 文件“cassandra/cluster.py”，第 2979 行，位于 cassandra.cluster.Session._create_response_future 文件“cassandra/query.py”，第 905 行，位于 cassandra.query.bind_params 类型错误：在字符串格式化期间并非所有参数都被转换

如果我将response_data设置为None。然后查询就可以正常执行了。但如果我尝试传递response_data，则会收到错误。 我做错了什么？

上面详细提到了

Answer 1

https://docs.datastax.com/en/developer/python-driver/3.28/user_defined_types/中的 Python 驱动程序文档解释了如何将用户定义类型作为绑定变量传递。该文档建议了一些选项，例如创建一个特殊的类，或“注册”用户定义的类型，但您也可以在没有类或注册的情况下使用命名元组来完成此操作。我不认为它只是天真地与字典一起工作，例如，

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)])