database_skill — 数据库技能

database-skill — Python 数据库连接技能 本技能指导 AI Agent 使用 Python 连接关系型数据库、执行 SQL 查询、管理事务以及探查 Schema 元数据。

何时调用 / 何时不调用 · 场景 调用 用户需要执行 SQL 查询 是 用户需要 Schema 探查（表、列、索引、外键） 是 用户需要事务控制（提交/回滚） 是 无 Python 3.8+ 或驱动未安装 否 用户仅需文本/正则分析 否

前提条件 · 要求 检查 Python 3.8+ python --version 目标数据库可达 telnet 依赖项 pip install pymysql psycopg2-binary oracledb pymssql pyyaml

快速开始 · 快速开始 # 列出所有表 python scripts/main.py \ --url "jdbc:mysql://localhost:3306/mydb" \ --user "root" \ --password "${DB_PASS}" \ --tables # 参数化查询 python scripts/main.py \ --url "jdbc:mysql://localhost:3306/mydb" \ --user "root" \ --password "${DB_PASS}" \ --query "SELECT FROM user WHERE name = ?" "zhangsan"

配置方式（三种） · 配置方式 A — 命令行参数（推荐） python scripts/main.py \ --url "jdbc:mysql://host:3306/db" \ --user "admin" \ --password "${DB_PASS}" \ --query "SELECT 1" B — YAML 配置文件 # datasource.yml datasource: url: "jdbc:mysql://localhost:3306/mydb" username: "${DB_USER}" password: "${DB_PASS}" python scripts/main.py --config datasource.yml --tables C — 内置默认 python scripts/main.py --tables

操作说明 · 操作 标志 说明 --query [params...] SELECT 查询，支持参数化 --update [params...] 更新/插入/删除 --batch 批量执行文件中的 SQL --tables 列出所有表 --columns 查看表结构 --list-connections 查看已保存的连接 --forget 删除已保存的连接

示例： python scripts/main.py --url "jdbc:mysql://host:3306/db" --user root --password x --tables python scripts/main.py --url "..." --user root --password x --query "SELECT FROM t WHERE id = ?" 42 python scripts/main.py --url "..." --user root --password x --update "UPDATE t SET x = ? WHERE id = ?" "new" 1 python scripts/main.py --list-connections

脚本结构 scripts/ ├── main.py # 命令行入口 ├── connection_manager.py # 连接管理 ├── connections_store.py # 连接记录持久化 ├── query_executor.py # 查询执行 ├── schema_inspector.py # Schema 探查 └── exceptions.py # 异常层次

API 概览 ConnectionManager · 连接管理器 from connection_manager import ConnectionManager cm = ConnectionManager(url="jdbc:mysql://localhost:3306/db", username="root", password="secret", driver="pymysql") conn = cm.get_connection() # auto-commit tx_conn = cm.get_connection_for_transaction() # manual-commit cm.close_connection(conn)

QueryExecutor · 查询执行器 from query_executor import QueryExecutor qe = QueryExecutor(cm) rows = qe.execute_query("SELECT * FROM users WHERE status = ?", "ACTIVE") affected = qe.execute_update("UPDATE users SET status = ? WHERE id = ?", "INACTIVE", 1) qe.execute_batch(["INSERT INTO log VALUES (1)", "INSERT INTO log VALUES (2)"]) qe.execute_transaction(lambda tx: ( tx.execute_update("UPDATE accounts SET balance = balance - 100 WHERE id = ?", 1), tx.execute_update("UPDATE accounts SET balance = balance + 100 WHERE id = ?", 2) ))

CaseInsensitiveDict · 大小写不敏感字典 查询结果的列名大小写不敏感： row["name"] == row["NAME"] == row["Name"]

SchemaInspector · Schema 检查器 from schema_inspector import SchemaInspector si = SchemaInspector(cm) si.get_tables() # 获取所有表 si.get_columns("users") # 获取表结构 si.get_indexes("users") # 获取索引 si.get_foreign_keys("users") # 获取外键

连接记录 连接成功后，技能会将 URL、用户名和驱动类型保存到用户临时目录。 密码不会被保存。

当不提供 --url 或 --config 时，显示已保存的连接供选择： Saved connections: [0] MySQL@host1 [1] Oracle@orahost [N] Enter a new connection python scripts/main.py --list-connections # 查看所有连接 python scripts/main.py --forget "jdbc:mysql://..." # 删除连接

安全守则 永远不要硬编码密码 — 使用环境变量 ${DB_PASS} 或命令行参数。 总是使用参数化查询 — 值