如何在Python代码中引用IntelliJ已配置的Cassandra数据源？

首先明确：IntelliJ的全局数据源配置是IDE内部存储的私有数据，Python的Cassandra驱动（比如官方的cassandra-driver）无法直接读取这些配置。你可以用以下几种常用方式复用这些参数，避免重复输入：

1. 使用环境变量存储配置 #

把IntelliJ里的Cassandra连接参数（用户名、密码、节点地址、端口、keyspace）设置为环境变量，Python代码直接读取这些变量：

示例代码： #

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
import os

# 从环境变量读取配置
username = os.getenv('CASSANDRA_USER')
password = os.getenv('CASSANDRA_PASSWORD')
contact_points = os.getenv('CASSANDRA_CONTACT_POINTS').split(',')
keyspace = os.getenv('CASSANDRA_KEYSPACE')

# 建立连接
auth_provider = PlainTextAuthProvider(username=username, password=password)
cluster = Cluster(contact_points=contact_points, auth_provider=auth_provider)
session = cluster.connect(keyspace)

# 执行查询示例
rows = session.execute("SELECT * FROM your_table LIMIT 5")
for row in rows:
    print(row)

# 关闭连接
cluster.shutdown()

你可以把这些环境变量添加到系统环境中，或者在PyCharm的Run Configuration里临时设置（Run > Edit Configurations > Environment variables）。

2. 使用本地配置文件 #

创建一个配置文件（比如.env或cassandra_config.json），把IntelliJ里的参数复制进去，Python代码读取这个文件：

示例：用.env文件（需安装python-dotenv库） #

先安装依赖：

pip install python-dotenv

创建.env文件：

CASSANDRA_USER=your_username
CASSANDRA_PASSWORD=your_password
CASSANDRA_CONTACT_POINTS=127.0.0.1
CASSANDRA_KEYSPACE=your_keyspace

Python代码读取：

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from dotenv import load_dotenv
import os

# 加载.env文件
load_dotenv()

# 读取配置
username = os.getenv('CASSANDRA_USER')
password = os.getenv('CASSANDRA_PASSWORD')
contact_points = [os.getenv('CASSANDRA_CONTACT_POINTS')]
keyspace = os.getenv('CASSANDRA_KEYSPACE')

# 建立连接并执行查询
auth_provider = PlainTextAuthProvider(username=username, password=password)
cluster = Cluster(contact_points=contact_points, auth_provider=auth_provider)
session = cluster.connect(keyspace)

# 示例查询
rows = session.execute("SELECT * FROM your_table LIMIT 5")
for row in rows:
    print(row)

cluster.shutdown()

3. 封装连接逻辑为工具函数 #

把连接Cassandra的代码封装成单独的工具函数，后续直接调用即可：

# cassandra_utils.py
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from dotenv import load_dotenv
import os

load_dotenv()

def get_cassandra_session():
    username = os.getenv('CASSANDRA_USER')
    password = os.getenv('CASSANDRA_PASSWORD')
    contact_points = [os.getenv('CASSANDRA_CONTACT_POINTS')]
    keyspace = os.getenv('CASSANDRA_KEYSPACE')
    
    auth_provider = PlainTextAuthProvider(username=username, password=password)
    cluster = Cluster(contact_points=contact_points, auth_provider=auth_provider)
    return cluster.connect(keyspace), cluster

业务代码中调用：

from cassandra_utils import get_cassandra_session

session, cluster = get_cassandra_session()
rows = session.execute("SELECT * FROM your_table LIMIT 5")
for row in rows:
    print(row)
cluster.shutdown()

内容的提问来源于stack exchange，提问作者PM 77-1