如果您使用的是DataStax's Python Driver for Cassandra (as of v3.28)，则无法使用Session.execute / Session.execute_async参数化或绑定表名，因为它只期望列值在parameters参数中，而表名在statement字符串中被硬编码。您将不得不求助于字符串格式，这会使您的链接器抱怨可能的SQL注入（然后您必须使用# noqa等方法使其静音）。
我通过使用cassandra.cqlengine.models来解决这个问题，它仍然是同一个Python驱动程序包的一部分：https://docs.datastax.com/en/developer/python-driver/3.28/api/cassandra/cqlengine/models/。
您可以在运行时动态地为表创建模型（在此期间，您可以从某处传递表名），然后使用该模型执行查询。
首先定义一个抽象模型：

from cassandra.cqlengine import columns
from cassandra.cqlengine.models import Model
from cassandra.cqlengine.management import sync_table

class MyAbstractModel(Model):
    __abstract__ = True

    __keyspace__ = "your_keyspace"

    # The table name will be defined during runtime
    # __table_name__ = ...

    id = columns.Integer(primary_key=True)
    first_name = columns.Text()
    last_name = columns.Text()

字符串
然后在运行时，创建一个具体模型：

def get_table_name() -> str:
    # Return some table name based on some app-specific logic.
    return "Person"

table_name = get_table_name()
table_attrs = {"__table_name__": table_name}
person_model = type(
    table_name,
    (MyAbstractModel,),
    table_attrs,
)

# Not shown here is setting-up the connection to the cluster
# and setting-up a session, as that's not part of the question.
# Assume an active session is already available.
sync_table(model=person_model)

型
type调用是Python内置的type函数，用于创建类型，在本例中是基于cassandra.cqlengine.models.Model的具体Person类。type的参数依次为：

• 类的名称（table_name）
• 基类（(MyAbstractModel,)）
• 模型的附加类属性（table_attrs），在本例中应为__table_name__（请参见https://docs.datastax.com/en/developer/python-driver/3.28/api/cassandra/cqlengine/models/）

如果代码成功运行，您将看到表已创建：

cqlsh> DESC TABLE your_keyspace.person;

CREATE TABLE your_keyspace.person (
    id int PRIMARY KEY,
    first_name text,
    last_name text
) WITH ...

cqlsh> SELECT COUNT(*) FROM your_keyspace.person;

 count
-------
     0

型
一旦你得到了你的表，你现在可以使用它来使你的查询 * 作为使用 * session.execute的替代。根据您需要创建的表的数量，您可能需要将表名Map到它们的模型：

all_models = {
    "person": person_model,
    "food": food_model,
    "drink": drink_model,
    ...
}

型
复制问题中的示例，其中表名是some_method的参数：

def some_method(self, table_name):
    model = all_models.get(table_name)
    if not model:
      raise InvalidTableNameException

    row = model(id=1234, first_name="Foo", last_name="Bar")
    row.save()

型
由于模型是动态创建的，因此只需确保

• 它仍然是一个有效的Python类名
• 它仍然是一个有效的Cassandra表名
• __table_name__和type名称一致

阅读Making Queries的文档和模型。