mcpField
C
cdata-cli-python

cdata-cli-python

github.com/cdatasoftware开发367 次安装1 周前更新
代码仓库

关于 cdata-cli-python

面向使用 CData Python Connector 的 Python 开发者的构建阶段技能,适用于 Python 3.x 应用。它将 cdata-cli 验证的连接与 SQL 延续到应用中,并介绍安装、授权及 DB-API 2.0 和 SQLAlchemy 用法。

主要功能

  1. 01获取并安装平台对应的 Python 连接器
  2. 02单独激活 Python 驱动许可证
  3. 03使用 DB-API 2.0 连接并执行参数化 SQL
  4. 04提供 SQLAlchemy 方言支持

适用场景

  1. 01将已验证的 SQL 用于 Python 应用
  2. 02在应用中复用发现阶段缓存的 OAuth 令牌

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/cdatasoftware/cli-skills --skill cdata-cli-python

安装后在 Claude Code 中输入/cdata-cli-python确认已加载。

详细介绍

技能:CData CLI — Python 构建

cdata-cli 技能的构建阶段配套技能。cdata-cli 使用 JDBC 驱动进行发现 (连接、探索架构、验证 SQL);开始编写 Python 应用代码并使用 CData Python Connector 后,则由本技能接手。

在发现阶段验证的内容都可以直接沿用:所有 CData 驱动共享相同的关系模型和 SQL 方言,因此,通过 cdatacli 确认的表名、列名和 SQL,在 Python Connector 中也无需更改即可使用。


语言 / 驱动版本

语言CData 版本Python 导入
Python 3.xCData Python Connectorimport cdata.<source>

<source> 是驱动名称的小写形式,例如:cdata.salesforce、cdata.gmail、 cdata.snowflake。连接器是一个编译后的 DB-API 2.0 模块(.pyd/.so),并附带 SQLAlchemy 方言,以平台专用 wheel 的形式发布。


前提条件:先完成发现

如果本次会话尚未使用 cdata-cli,请停止并先调用它。 本技能仅涵盖构建阶段——继续操作前,必须通过 cdata-cli 完成连接设置、架构发现和 SQL 验证。

本技能假定已使用 cdata-cli 完成以下操作:

  • 确认数据源可以连接;以及
  • 验证应用将要运行的确切表、列和 SQL。

在此处重用这些结果。特别是,如果发现阶段创建的 cdatacli 连接执行过 OAuth 浏览器登录,请记下其 OAuthSettingsLocation——Python 应用可以指向同一缓存令牌文件,从而跳过重新认证(参见 连接)。

运行时方面,CLI 仅直接使用 Java/JDBC。 它无法为 Python Connector 安装、激活或创建连接——这些正是本技能涵盖的内容。Python Connector 是独立的运行时驱动版本,必须单独获取并授权(步骤如下)。

CLI 的 drivers download 只会获取 JDBC jar;Python wheel 则通过普通 HTTP 从 CData 的 Python 仓库获取(见下文)。


步骤 1:获取连接器 Wheel

CData Python Connectors 以可浏览的索引形式发布在 https://maven.cdata.com/python/(普通 HTTP——无 API、无需认证)。

  1. 查找制品:cdata-<source>-connector(例如 cdata-salesforce-connector)。
  2. 打开最新版本的文件夹(例如 26.0.9655)。
  3. 选择与目标操作系统 / 架构匹配的 wheel:
平台Wheel 后缀
Windows x64...-win_amd64.whl
Linux x64...-linux_x86_64.whl
macOS Apple Silicon...-macosx_<ver>_arm64.whl

cp310-abi3 标签表示支持 Python 3.10+ 和稳定 ABI——一个 wheel 即可适用于所有 Python ≥ 3.10 版本。

下载并验证文件(服务器偶尔会在传输中途断开连接):

# bash
curl -fSL --retry 3 --retry-all-errors -o cdata_<source>_connector-<ver>-cp310-abi3-win_amd64.whl \
  "https://maven.cdata.com/p/python/cdata-<source>-connector/<ver>/cdata_<source>_connector-<ver>-cp310-abi3-win_amd64.whl"
# PowerShell — curl.exe is more robust here than Invoke-WebRequest
curl.exe -fSL --retry 3 --retry-all-errors -o "cdata_<source>_connector-<ver>-cp310-abi3-win_amd64.whl" `
  "https://maven.cdata.com/p/python/cdata-<source>-connector/<ver>/cdata_<source>_connector-<ver>-cp310-abi3-win_amd64.whl"

.whl 是一个 zip 文件;如果无法作为归档文件打开,说明文件被截断——请重新下载。

本地已安装? 如果通过 CData 产品套件安装了连接器,可以使用其中的 wheel,无需下载——请在解压路径下查找:

<install-path>\CData.Python.<Source>\win\Python<ver>\64\cdata_<source>_connector-...whl   # Windows
<install-path>/CData.Python.<Source>/linux/cdata_<source>_connector-...tar.gz             # Linux/macOS

步骤 2:安装连接器

将 wheel 安装到应用将要运行的解释器中:

python -m pip install cdata_<source>_connector-<ver>-cp310-abi3-win_amd64.whl

建议使用虚拟环境(可将安装限制在项目内,并使许可证工具路径保持可预测),但并非必需——全局安装的效果相同:

python -m venv .venv    # optional; activate, then pip install into it

验证模块是否可以导入:

python -c "import cdata.<source> as m; print('paramstyle:', m.paramstyle, '| apilevel:', m.apilevel)"
# expect: paramstyle: qmark | apilevel: 2.0

步骤 3:激活许可证

cdatacli 连接正常并不代表许可证已激活。 cdatacli 基于 JDBC 驱动运行,因此只能证明 JDBC 已获许可——Python Connector 使用单独的许可证。

Python Connector 的许可证与 JDBC/ADO.NET/ODBC 版本分开授权,并且按机器授权。未激活时,首次查询会报错:"Could not find a valid license…"。

激活程序随已安装的软件包一同提供,位于当前解释器的 site-packages/cdata/installlic_<source>/ 中(使用以下命令查找): python -c "import cdata, os; print(os.path.dirname(cdata.__file__))":

installlic_<source>/install-license.exe   # Windows
installlic_<source>/install-license.sh    # macOS/Linux

请在激活程序所在文件夹中运行它(它会从当前目录读取 prod.inf)。传入 --name/--email 可进行非交互式运行(build 26.0.9704+;较早的 build 会显示提示,因此用户必须在普通终端中运行)。在 macOS/Linux 上使用 ./install-license.sh:

cd <installlic_<source> folder>       # path located above
./install-license.exe --name "First Last" --email [email protected]               # trial — safe to run here
./install-license.exe --name "First Last" --email [email protected] <PRODUCT-KEY> # key is a secret — user runs outside the AI session

激活后,重新运行导入/查询检查以确认激活生效。


步骤 4:连接(DB-API 2.0)

import cdata.<source> as mod

conn = mod.connect(CONNECTION_STRING)   # a full CData connection string
cur = conn.cursor()
cur.execute("SELECT Id, Name FROM Account WHERE Name LIKE ?", ["%Acme%"])
rows = cur.fetchall()
  • paramstyle 为 qmark——使用 ? 占位符,并向 execute 传入列表。始终对用户输入使用参数化处理;切勿使用字符串格式化 SQL。
  • 执行 INSERT/UPDATE/DELETE 后调用 conn.commit()。
连接字符串与身份验证(复用发现阶段的配置,勿将密钥写入代码)

传入已使用 cdatacli 验证过的相同属性。对于 OAuth 数据源,将 OAuthSettingsLocation 指向发现阶段的连接已填充的令牌文件,以便应用静默刷新,无需浏览器授权流程:

# Embedded-OAuth source (e.g. Salesforce) — cached token is sufficient:
cs = ("AuthScheme=OAuth;InitiateOAuth=GETANDREFRESH;UseSandbox=true;"
      r"OAuthSettingsLocation=C:\...\CData\Salesforce Data Provider\OAuthSettings-xxx.txt")

步骤 5:端到端示例

与框架无关(根据需要封装在 Flask/FastAPI/Tkinter 等框架中):

import cdata.salesforce as sf

CS = ("AuthScheme=OAuth;InitiateOAuth=GETANDREFRESH;UseSandbox=true;"
      r"OAuthSettingsLocation=C:\...\OAuthSettings-sbx.txt")

def get_conn():
    return sf.connect(CS)

def list_accounts(search=None, limit=50):
    conn = get_conn()
    try:
        cur = conn.cursor()
        sql = "SELECT Id, Name, Industry FROM Account"
        params = []
        if search:
            sql += " WHERE Name LIKE ?"; params.append(f"%{search}%")
        sql += f" ORDER BY Name LIMIT {int(limit)}"
        cur.execute(sql, params)
        return cur.fetchall()
    finally:
        conn.close()

def create_account(name, industry=None):
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("INSERT INTO Account (Name, Industry) VALUES (?, ?)", [name, industry])
        conn.commit()
    finally:
        conn.close()

if __name__ == "__main__":
    for row in list_accounts(limit=5):
        print(row)

SQLAlchemy(可选)

此 wheel 也会安装 SQLAlchemy 方言,因此 ORM/引擎代码也可正常运行:

  • cdata.sqlalchemy_<source>——SQLAlchemy 1.4 风格的方言
  • cdata.sqlalchemy2_<source>——SQLAlchemy 2.0 风格的方言
from sqlalchemy import create_engine, text
# URL form: <source>:///?<connection-string-properties>
engine = create_engine("salesforce:///?AuthScheme=OAuth;InitiateOAuth=GETANDREFRESH;UseSandbox=true")
with engine.connect() as c:
    print(c.execute(text("SELECT COUNT(*) FROM Account")).scalar())

故障排除

错误原因 / 解决方法
Could not find a valid license for using CData Python Driver for <source>此机器上的 Python Connector 尚未激活——运行 install-license(步骤 3)。它与任何 JDBC/ADO.NET/ODBC 许可证相互独立。
'OAuthClientId' and 'OAuthClientSecret' are needed to initiate OAuth数据源未嵌入 OAuth 应用;在连接字符串中提供客户端 ID/密钥(从环境变量读取)。必须与缓存了刷新令牌的应用匹配。
object does not support the context manager protocol使用了 with mod.connect(...)。请改用显式的 try/finally。
ModuleNotFoundError: No module named 'cdata.<source>'当前解释器中未安装 wheel——确认 venv 已激活 / wheel 已安装,然后重新检查 import cdata.<source>。
install-license.exe hangs or errors on prod.inf请在激活程序所在文件夹中运行它,并传入 --name/--email(步骤 3);若不传入,它会停在提示处等待输入。早于 26.0.9704 的 build 不支持这些标志——请升级,或让用户以交互方式运行。

评论

支持 Markdown