在 Cursor 中操作本地数据库（如 SQLite、MySQL、PostgreSQL），核心是借助 Cursor 的 MCP 协议（本地资源访问）+ AI 辅助编码，无需手动写复杂连接逻辑和 SQL 语句，全程可视化且高效。以下是通用步骤（以 SQLite 为例，MySQL/PostgreSQL 仅需微调依赖和连接参数）：

前置准备

1. 环境：Cursor 已打开本地项目文件夹（数据库文件需在项目目录下，或知道绝对路径）；
2. 数据库依赖：根据数据库类型安装对应 Python 库（Cursor 会帮你生成安装命令）：
   • SQLite：内置 sqlite3 库（无需额外安装）；
   • MySQL：需安装 pymysql 或 mysql-connector-python；
   • PostgreSQL：需安装 psycopg2-binary。

核心步骤（5 步搞定：连接→操作→验证）

步骤 1：让 AI 生成数据库连接代码（Cursor 自动适配本地环境）

Cursor 的 MCP 能识别本地项目路径，AI 会帮你生成 “连接本地数据库” 的通用代码，无需手动写连接参数。

操作：

1. 在 Cursor 中新建文件 db_operation.py；
2. 在文件中输入提示词（明确数据库类型、路径、需求），按 Alt+\ 让 AI 生成代码：

   # 提示词：帮我写 Python 代码，连接本地 SQLite 数据库（路径：./data/user.db，若不存在则自动创建）
   # 要求：1. 建立数据库连接；2. 创建 User 表（字段：id 主键自增，username 唯一，age 整数，create_time 时间戳）；
   # 3. 封装连接函数，返回数据库连接对象；4. 包含异常处理（如连接失败提示）
   

生成的代码（直接可用，Cursor 会适配本地路径）：

import sqlite3
import datetime
from typing import Optional

def get_db_connection(db_path: str = "./data/user.db") -> Optional[sqlite3.Connection]:
    """
    连接本地 SQLite 数据库，不存在则自动创建
    :param db_path: 数据库文件路径（相对项目根目录）
    :return: 数据库连接对象，失败返回 None
    """
    try:
        # 建立连接（MCP 会确保路径是本地项目内的有效路径）
        conn = sqlite3.connect(db_path, check_same_thread=False)
        # 设置行工厂，查询结果以字典形式返回（更易用）
        conn.row_factory = sqlite3.Row
        print(f"✅ 成功连接数据库：{db_path}")
        return conn
    except Exception as e:
        print(f"❌ 数据库连接失败：{str(e)}")
        return None

def create_user_table(conn: sqlite3.Connection):
    """创建 User 表（若不存在）"""
    try:
        cursor = conn.cursor()
        create_sql = """
        CREATE TABLE IF NOT EXISTS User (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT NOT NULL UNIQUE,
            age INTEGER,
            create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );
        """
        cursor.execute(create_sql)
        conn.commit()
        print("✅ User 表创建成功（或已存在）")
    except Exception as e:
        conn.rollback()
        print(f"❌ 创建 User 表失败：{str(e)}")

# 初始化数据库（连接+建表）
if __name__ == "__main__":
    db_conn = get_db_connection()
    if db_conn:
        create_user_table(db_conn)
        db_conn.close()  # 初始化完成后关闭连接

关键说明：

• 若用 MySQL，提示词中补充 “使用 pymysql 库，本地 MySQL 地址 127.0.0.1，端口 3306，用户名 root，密码 123456，数据库名 test_db”，AI 会自动生成 MySQL 连接代码；
• Cursor 会通过 MCP 验证本地路径有效性，避免路径错误。

步骤 2：用 AI 生成 CRUD 操作代码（增删改查）

无需手动写 SQL，用自然语言描述需求，AI 会基于已有的连接函数，生成适配的 CRUD 逻辑。

操作：

在 db_operation.py 中继续输入提示词，让 AI 追加代码：

# 提示词：基于上面的数据库连接和 User 表，补充 4 个函数：
# 1. add_user：添加用户（参数：username, age），返回添加成功的用户 ID；
# 2. get_user_by_username：根据用户名查询用户，返回用户信息字典；
# 3. update_user_age：根据用户名修改年龄，返回是否修改成功；
# 4. delete_user：根据用户名删除用户，返回是否删除成功；
# 要求：包含参数校验、异常处理、事务提交/回滚

生成的代码（直接追加到文件中）：

def add_user(conn: sqlite3.Connection, username: str, age: int) -> Optional[int]:
    """添加用户"""
    if not username or age < 0:
        print("❌ 用户名不能为空，年龄不能为负数")
        return None
    try:
        cursor = conn.cursor()
        insert_sql = "INSERT INTO User (username, age) VALUES (?, ?)"
        cursor.execute(insert_sql, (username, age))
        conn.commit()
        print(f"✅ 添加用户成功：{username}（ID：{cursor.lastrowid}）")
        return cursor.lastrowid
    except sqlite3.IntegrityError:
        conn.rollback()
        print(f"❌ 用户名 {username} 已存在")
        return None
    except Exception as e:
        conn.rollback()
        print(f"❌ 添加用户失败：{str(e)}")
        return None

def get_user_by_username(conn: sqlite3.Connection, username: str) -> Optional[dict]:
    """根据用户名查询用户"""
    try:
        cursor = conn.cursor()
        query_sql = "SELECT * FROM User WHERE username = ?"
        cursor.execute(query_sql, (username,))
        row = cursor.fetchone()
        if row:
            # 转换为字典返回
            return dict(row)
        else:
            print(f"❌ 未找到用户：{username}")
            return None
    except Exception as e:
        print(f"❌ 查询用户失败：{str(e)}")
        return None

def update_user_age(conn: sqlite3.Connection, username: str, new_age: int) -> bool:
    """修改用户年龄"""
    if new_age < 0:
        print("❌ 年龄不能为负数")
        return False
    try:
        cursor = conn.cursor()
        update_sql = "UPDATE User SET age = ? WHERE username = ?"
        cursor.execute(update_sql, (new_age, username))
        conn.commit()
        if cursor.rowcount > 0:
            print(f"✅ 已将 {username} 的年龄修改为 {new_age}")
            return True
        else:
            print(f"❌ 未找到用户：{username}（修改失败）")
            return False
    except Exception as e:
        conn.rollback()
        print(f"❌ 修改用户年龄失败：{str(e)}")
        return False

def delete_user(conn: sqlite3.Connection, username: str) -> bool:
    """删除用户"""
    try:
        cursor = conn.cursor()
        delete_sql = "DELETE FROM User WHERE username = ?"
        cursor.execute(delete_sql, (username,))
        conn.commit()
        if cursor.rowcount > 0:
            print(f"✅ 已删除用户：{username}")
            return True
        else:
            print(f"❌ 未找到用户：{username}（删除失败）")
            return False
    except Exception as e:
        conn.rollback()
        print(f"❌ 删除用户失败：{str(e)}")
        return False

步骤 3：本地测试数据库操作（Cursor 终端执行）

Cursor 内置终端，可直接运行代码测试，AI 会帮你生成测试脚本，无需切换其他工具。

操作：

1. 在 db_operation.py 末尾添加测试代码（让 AI 生成）：

   # 提示词：写一个测试函数，依次执行：连接数据库→添加2个用户→查询用户→修改年龄→删除用户→关闭连接
   def test_db_operations():
       # 1. 连接数据库
       conn = get_db_connection()
       if not conn:
           return
       
       try:
           # 2. 添加用户
           add_user(conn, "zhangsan", 25)
           add_user(conn, "lisi", 30)
           
           # 3. 查询用户
           user = get_user_by_username(conn, "zhangsan")
           if user:
               print("📋 查询结果：", user)
           
           # 4. 修改年龄
           update_user_age(conn, "zhangsan", 26)
           
           # 5. 删除用户
           delete_user(conn, "lisi")
           
           # 6. 再次查询验证
           print("📋 最终用户列表：")
           cursor = conn.cursor()
           cursor.execute("SELECT * FROM User")
           for row in cursor.fetchall():
               print(dict(row))
       finally:
           # 7. 关闭连接
           conn.close()
           print("🔌 数据库连接已关闭")
   

执行测试

if name == "main":

初始化表（之前的代码）

db_conn = get_db_connection()if db_conn:create_user_table(db_conn)db_conn.close()

执行测试

test_db_operations()

2. 打开 Cursor 内置终端（`Ctrl+`` 或点击底部“Terminal”）；
3. 运行代码（自动安装依赖，若有缺失）：
   ```bash
   # SQLite 无需安装依赖，直接运行
   python db_operation.py

预期输出（终端中）：

✅ 成功连接数据库：./data/user.db
✅ User 表创建成功（或已存在）
✅ 成功连接数据库：./data/user.db
✅ 添加用户成功：zhangsan（ID：1）
✅ 添加用户成功：lisi（ID：2）
📋 查询结果： {'id': 1, 'username': 'zhangsan', 'age': 25, 'create_time': '2025-11-14 15:30:00'}
✅ 已将 zhangsan 的年龄修改为 26
✅ 已删除用户：lisi
📋 最终用户列表：
{'id': 1, 'username': 'zhangsan', 'age': 26, 'create_time': '2025-11-14 15:30:00'}
🔌 数据库连接已关闭

步骤 4：按需扩展复杂操作（AI 辅助写 SQL）

若需要复杂查询（如分页、联表查询、统计），直接用自然语言描述需求，让 AI 生成代码，无需手动写 SQL。

示例场景：分页查询用户 + 统计总人数

操作：

在文件中输入提示词：

# 提示词：补充 2 个函数：
# 1. get_users_by_page：分页查询用户（参数：page 页码，page_size 每页条数），返回用户列表和总条数；
# 2. count_total_users：统计 User 表总用户数；
# 基于现有数据库连接逻辑，包含异常处理

生成的代码：

def get_users_by_page(conn: sqlite3.Connection, page: int = 1, page_size: int = 10) -> tuple[list[dict], int]:
    """分页查询用户"""
    if page < 1 or page_size < 1:
        print("❌ 页码和每页条数必须为正整数")
        return [], 0
    try:
        total = count_total_users(conn)
        offset = (page - 1) * page_size
        cursor = conn.cursor()
        query_sql = "SELECT * FROM User LIMIT ? OFFSET ?"
        cursor.execute(query_sql, (page_size, offset))
        rows = cursor.fetchall()
        user_list = [dict(row) for row in rows]
        print(f"✅ 第 {page} 页查询完成（共 {total} 条，当前页 {len(user_list)} 条）")
        return user_list, total
    except Exception as e:
        print(f"❌ 分页查询失败：{str(e)}")
        return [], 0

def count_total_users(conn: sqlite3.Connection) -> int:
    """统计总用户数"""
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT COUNT(*) AS total FROM User")
        row = cursor.fetchone()
        return row["total"] if row else 0
    except Exception as e:
        print(f"❌ 统计用户数失败：{str(e)}")
        return 0

步骤 5：调试与优化（Cursor AI 帮你排查问题）

若运行中出现报错（如路径错误、SQL 语法错误、权限问题），无需手动排查，让 Cursor AI 帮你解决：

常见问题处理：

1. 数据库路径错误：终端提示 “找不到文件”，选中报错信息，在 AI 侧边栏输入：“帮我修复这个数据库路径错误，确保代码能找到 ./data/user.db 文件”，AI 会自动修改 get_db_connection 函数，补充路径校验或自动创建 data 文件夹；
2. MySQL 连接失败：提示 “Access denied”，输入提示词：“我用 MySQL 数据库，本地地址 127.0.0.1，端口 3306，用户名 root，密码 123456，数据库名 test_db，帮我修改连接函数，使用 pymysql 库，解决连接失败问题”；
3. SQL 语法错误：选中报错的 SQL 相关代码，问 AI：“这个 SQL 语句报错，帮我修复并解释原因”。

不同数据库的适配技巧（MySQL/PostgreSQL）

1. MySQL 适配（仅需修改 2 处）

（1）安装依赖：

在终端运行 AI 生成的命令：

pip install pymysql

（2）修改连接函数（让 AI 生成）：

输入提示词：“将上面的 SQLite 连接函数改为 MySQL 连接，本地 MySQL 地址 127.0.0.1，端口 3306，用户名 root，密码 123456，数据库名 test_db，使用 pymysql 库”，AI 会生成：

import pymysql
from typing import Optional

def get_mysql_connection() -> Optional[pymysql.connections.Connection]:
    try:
        conn = pymysql.connect(
            host="127.0.0.1",
            port=3306,
            user="root",
            password="123456",
            database="test_db",
            charset="utf8mb4",
            cursorclass=pymysql.cursors.DictCursor  # 查询结果返回字典
        )
        print("✅ MySQL 连接成功")
        return conn
    except Exception as e:
        print(f"❌ MySQL 连接失败：{str(e)}")
        return None

2. PostgreSQL 适配

（1）安装依赖：

pip install psycopg2-binary

（2）修改连接函数（AI 生成）：

提示词：“改为 PostgreSQL 连接，本地地址 127.0.0.1，端口 5432，用户名 postgres，密码 123456，数据库名 test_db，使用 psycopg2 库”，AI 会自动生成对应代码。

核心总结

在 Cursor 中操作本地数据库的核心逻辑是：AI 帮你写代码，MCP 保障本地资源访问，终端直接测试，无需手动处理连接、SQL、异常等繁琐工作，步骤可简化为：

1. 明确数据库类型 + 需求（用自然语言描述）；
2. 让 AI 生成 “连接 + 建表 + CRUD” 代码；
3. 用 Cursor 终端运行测试；
4. 报错时让 AI 排查修复。

关键技巧：提示词越具体，生成的代码越精准，比如明确 “数据库类型、连接参数、字段约束、操作逻辑”，AI 能直接生成可运行的代码，无需二次修改。