在 Python 中操作 SQL Server，最主流、最推荐使用的库是 pyodbc，它是一个功能强大、稳定且广泛支持的数据库连接库，另一个选择是 pymssql，它更轻量，但功能相对较少。

本指南将主要围绕 pyodbc 展开，因为它提供了更接近原生 ODBC 驱动的性能和兼容性。

第一步：环境准备

在开始编写 Python 代码之前，你需要完成以下几个关键步骤：

安装 Python ODBC 驱动

这是最关键的一步,Python 的 pyodbc 库本身并不包含数据库驱动的代码，它需要调用操作系统级别的 ODBC 驱动程序来与 SQL Server 通信。

你需要安装 Microsoft ODBC Driver for SQL Server。

• 下载地址: Microsoft ODBC Driver for SQL Server
• 选择版本: 根据你的操作系统和 SQL Server 版本选择合适的驱动，对于大多数现代应用，推荐 ODBC Driver 18 for SQL Server，安装时，确保勾选所有组件。

安装 Python pyodbc 库

使用 pip 来安装 pyodbc。

pip install pyodbc

连接信息准备

你需要准备好连接到你的 SQL Server 实例所需的信息：

• 服务器名称: localhost, 0.0.1, 或一个远程服务器的地址。
• 数据库名称: 你要连接的数据库。
• 用户名和密码: 用于身份验证的凭据。
• 认证方式: SQL Server 支持多种认证方式，最常见的是：
  • Windows 身份验证: 使用你的 Windows 账户登录，无需用户名密码。
  • SQL Server 身份验证: 需要明确的用户名和密码。

第二步：建立数据库连接

连接是所有操作的基础。pyodbc 使用连接字符串来指定如何连接到数据库。

连接字符串模板

连接字符串的格式如下, 中的部分需要根据你的实际情况替换。

# 使用 SQL Server 身份验证
DRIVER = '{ODBC Driver 18 for SQL Server}'  # 驱动名称，必须与安装的驱动完全一致
SERVER = 'your_server_name'
DATABASE = 'your_database_name'
USERNAME = 'your_username'
PASSWORD = 'your_password'
connection_string = f"DRIVER={DRIVER};SERVER={SERVER};DATABASE={DATABASE};UID={USERNAME};PWD={PASSWORD}"
# 使用 Windows 身份验证 (更安全，推荐在内部网络使用)
# connection_string = f"DRIVER={DRIVER};SERVER={SERVER};DATABASE={DATABASE};Trusted_Connection=yes;"

注意: 驱动名称 {ODBC Driver 18 for SQL Server} 必须与你安装的驱动名称完全匹配，包括大写、空格和版本号，你可以通过 ODBC 数据源管理器或在代码中尝试不同名称来确认。

建立和关闭连接

import pyodbc
def get_connection():
    """建立并返回一个数据库连接"""
    try:
        conn = pyodbc.connect(connection_string)
        print("连接成功！")
        return conn
    except pyodbc.Error as e:
        print(f"连接失败: {e}")
        return None
# 使用示例
conn = get_connection()
if conn:
    # 在这里执行数据库操作...
    pass
# 操作完成后，务必关闭连接
if conn:
    conn.close()
    print("连接已关闭。")

第三步：执行 SQL 语句

连接成功后,你可以使用 cursor 对象来执行 SQL 语句，主要有两种操作：查询 和 非查询。

执行查询操作

查询操作会返回数据,你需要使用 fetchone(), fetchall() 或 fetchmany() 来获取结果。

import pyodbc
# 假设 conn 已经成功建立
try:
    # 1. 创建一个 cursor 对象
    cursor = conn.cursor()
    # 2. 编写 SQL 查询语句
    sql_query = "SELECT TOP 10 * FROM YourTableName;"
    # 3. 执行 SQL 语句
    cursor.execute(sql_query)
    # 4. 获取查询结果
    # fetchone(): 获取下一行结果
    # row = cursor.fetchone()
    # if row:
    #     print(row)
    # fetchall(): 获取所有剩余的行
    rows = cursor.fetchall()
    # 5. 遍历并打印结果
    print("查询结果:")
    for row in rows:
        # row 是一个类似元组的对象，可以通过索引或列名访问
        # print(row) # 打印整行
        print(f"ID: {row[0]}, Name: {row[1]}") # 假设第一列是ID，第二列是Name
except pyodbc.Error as e:
    print(f"查询出错: {e}")
finally:
    # 6. 关闭 cursor
    if 'cursor' in locals():
        cursor.close()

执行非查询操作

非查询操作包括 INSERT, UPDATE, DELETE 等，这些操作会修改数据库，因此需要 提交事务 (commit()) 才能使更改永久生效。

import pyodbc
# 假设 conn 已经成功建立
try:
    cursor = conn.cursor()
    # --- 插入数据 ---
    insert_sql = "INSERT INTO YourTableName (Name, Age) VALUES (?, ?);"
    # 使用 ? 作为参数占位符，防止 SQL 注入！
    data_to_insert = ('Alice', 30)
    cursor.execute(insert_sql, data_to_insert)
    # --- 更新数据 ---
    update_sql = "UPDATE YourTableName SET Age = ? WHERE Name = ?;"
    data_to_update = (31, 'Alice')
    cursor.execute(update_sql, *data_to_update) # 使用 * 解包元组
    # --- 删除数据 ---
    delete_sql = "DELETE FROM YourTableName WHERE Name = ?;"
    name_to_delete = ('Bob',)
    cursor.execute(delete_sql, name_to_delete)
    # 提交事务，这是至关重要的一步！
    conn.commit()
    print(f"成功执行了 {cursor.rowcount} 条修改操作。")
except pyodbc.Error as e:
    # 如果发生错误，回滚事务
    conn.rollback()
    print(f"操作出错，已回滚: {e}")
finally:
    if 'cursor' in locals():
        cursor.close()

第四步：参数化查询（防止 SQL 注入）

在上面的例子中,你已经看到了参数化查询的使用，这是编写安全数据库代码的 黄金法则。

永远不要使用 Python 字符串拼接来构建 SQL 查询！

❌ 危险的做法（易受 SQL 注入攻击）

user_input = "admin' -- "
# 恶意用户输入这个，可能会导致删除整个表
sql = f"SELECT * FROM Users WHERE username = '{user_input}'"
# 实际执行的 SQL 可能是: SELECT * FROM Users WHERE username = 'admin' -- '
# -- 是 SQL 注释，后面的内容会被忽略，可能导致绕过密码验证

✅ 安全的做法（使用参数化查询）

user_input = "admin' -- "
# 使用 ? 作为占位符
sql = "SELECT * FROM Users WHERE username = ?;"
# 将参数作为第二个参数传递给 execute
cursor.execute(sql, user_input)
# pyodbc 会安全地处理参数，将其作为数据处理，而不是 SQL 代码的一部分。

第五步：使用 with 语句（最佳实践）

手动管理 cursor 和 connection 的关闭容易出错，推荐使用 with 语句来自动处理资源的释放。

import pyodbc
# 连接字符串
connection_string = "your_connection_string_here"
# 使用 with 语句管理连接
with pyodbc.connect(connection_string) as conn:
    print("连接已建立")
    # 使用 with 语句管理 cursor
    with conn.cursor() as cursor:
        sql = "SELECT GETDATE() AS CurrentDateTime;"
        cursor.execute(sql)
        row = cursor.fetchone()
        if row:
            print(f"服务器当前时间: {row.CurrentDateTime}") # 通过列名访问
    # 当离开 with 块时，cursor 会自动关闭
# 当离开这个 with 块时，conn 会自动关闭
print("连接已自动关闭。")

第六步：将查询结果转换为 Pandas DataFrame

数据分析中,一个常见的需求是将查询结果直接加载到 Pandas DataFrame 中，这非常简单。

安装 Pandas

pip install pandas

使用 read_sql_query 或 read_sql_table

import pyodbc
import pandas as pd
connection_string = "your_connection_string_here"
try:
    with pyodbc.connect(connection_string) as conn:
        # 方法一: 使用 SQL 查询
        sql = "SELECT TOP 5 * FROM YourTableName;"
        df_query = pd.read_sql_query(sql, conn)
        print("通过 read_sql_query 生成的 DataFrame:")
        print(df_query)
        # 方法二: 直接读取整个表 (如果不需要复杂查询)
        # table_name = "YourTableName"
        # df_table = pd.read_sql_table(table_name, conn)
        # print("\n通过 read_sql_table 生成的 DataFrame:")
        # print(df_table)
except pyodbc.Error as e:
    print(f"数据库操作失败: {e}")

常见问题与解决方案

1. 错误: Data source name not found and no default driver specified

   • 原因: ODBC 驱动未安装，或者连接字符串中的 DRIVER 名称不正确。
   • 解决: 确保已正确安装 Microsoft ODBC Driver，并检查连接字符串中的驱动名称是否与 ODBC 驱动管理器中显示的完全一致。

2. 错误: Login failed for user '...'

   • 原因: 用户名或密码错误，或者用户没有权限访问该数据库/服务器。
   • 解决: 检查凭据是否正确，并确认该用户在 SQL Server 中有相应的登录权限和数据库访问权限。

3. 错误: Communication link failure

   • 原因: 网络问题，或者 SQL Server 的 TCP/IP 协议未启用。
   • 解决:
     • 确保你的机器可以 ping 通 SQL Server 的地址。
     • 在 SQL Server Configuration Manager 中，确保 SQL Server Network Configuration 下的 TCP/IP 协议是 Enabled 状态。

遵循以上步骤和最佳实践,你就可以在 Python 中安全、高效地操作 SQL Server 数据库了。