核心思想

在 Python 中操作任何数据库,基本流程都是一样的：

1. 安装驱动：安装一个允许 Python 与 SQL Server 通信的库。
2. 建立连接：使用驱动提供的 API,创建一个到数据库服务器的连接对象。
3. 创建游标：通过连接对象创建一个游标，游标用于执行 SQL 语句和获取结果。
4. 执行命令：使用游标执行 SELECT（查询）、INSERT（插入）、UPDATE（更新）、DELETE（删除）等 SQL 语句。
5. 处理结果：如果是查询语句，从游标中获取数据；如果是增删改语句,提交事务以使更改永久化。
6. 关闭资源：关闭游标和连接,释放资源。

环境准备：安装驱动库

目前最主流、性能最好的库是 pyodbc,微软官方也推荐使用它。

安装 pyodbc

打开你的终端或命令提示符,运行以下命令：

pip install pyodbc

系统依赖（重要！）

pyodbc 本身只是一个接口，它需要调用系统上的 ODBC 驱动程序来与 SQL Server 通信，你必须在你的操作系统上安装相应的 ODBC 驱动。

对于 Windows:

1. 访问微软官方 ODBC 驱动下载页面：Microsoft ODBC Driver for SQL Server
2. 下载并安装最新版本的 "ODBC Driver 17 for SQL Server" 或 "ODBC Driver 18 for SQL Server",安装程序会自动配置好系统。

对于 Linux (以 Ubuntu 为例):

# 更新包列表
sudo apt-get update
# 安装 ODBC 驱动
sudo apt-get install -y msodbcsql17
# 可能还需要安装 unixODBC 开发头文件
sudo apt-get install -y unixodbc-dev

对于 macOS (使用 Homebrew):

# 添加微软的 tap
brew tap microsoft/mssql-release
# 安装 ODBC 驱动
brew install --no-sandbox msodbcsql18
# 安装 unixODBC (如果尚未安装)
brew install unixodbc

基本操作：连接、查询、插入

1 建立连接

连接字符串是连接数据库的关键，它包含了服务器地址、数据库名、认证方式等信息。

连接字符串格式: DRIVER={...};SERVER=...;DATABASE=...;UID=...;PWD=...

• DRIVER: ODBC 驱动的名称，必须与你安装的驱动版本匹配。
  • ODBC Driver 17 for SQL Server
  • ODBC Driver 18 for SQL Server
• SERVER: SQL Server 的地址，可以是服务器名、IP 地址或 服务器名\实例名。
• DATABASE: 要连接的数据库名称。
• UID: 用户名。
• PWD: 密码。

示例代码:

import pyodbc
# --- 1. 定义连接字符串 ---
# 请根据你的实际情况修改以下信息
server = 'your_server_name'  # 'localhost', '192.168.1.100', 'SERVERNAME\INSTANCE'
database = 'your_database_name'
username = 'your_username'
password = 'your_password'
driver = '{ODBC Driver 17 for SQL Server}' # 根据你安装的驱动版本修改
# 构建完整的连接字符串
connection_string = f'DRIVER={driver};SERVER={server};DATABASE={database};UID={username};PWD={password}'
try:
    # --- 2. 建立连接 ---
    conn = pyodbc.connect(connection_string)
    print("连接成功！")
    # --- 3. 创建游标 ---
    cursor = conn.cursor()
    # --- 4. 执行查询 ---
    sql_query = "SELECT * FROM YourTableName WHERE ID = ?" # 使用参数化查询防止SQL注入
    cursor.execute(sql_query, 1) # 传递参数
    # --- 5. 处理结果 ---
    row = cursor.fetchone()
    if row:
        print(f"查询结果: ID={row[0]}, Name={row[1]}, Email={row[2]}")
    else:
        print("未找到记录")
except pyodbc.Error as e:
    print(f"数据库操作失败: {e}")
finally:
    # --- 6. 关闭资源 ---
    if 'conn' in locals() and conn:
        cursor.close()
        conn.close()
        print("连接已关闭。")

增、删、改 操作

对于 INSERT, UPDATE, DELETE 操作，执行完毕后需要提交事务,否则更改不会保存到数据库中。

示例：插入数据

重要：使用参数化查询！ 直接将变量拼接到 SQL 语句中是极其危险的，会导致 SQL 注入攻击，务必使用 作为占位符，并将参数作为第二个参数传递给 execute() 方法。

import pyodbc
# (使用与上面相同的连接字符串和连接逻辑)
# ... (连接代码省略，假设 conn 已建立) ...
try:
    cursor = conn.cursor()
    # 1. 准备 SQL 语句和参数
    sql_insert = "INSERT INTO Employees (Name, Email, HireDate) VALUES (?, ?, ?)"
    params = ('张三', 'zhangsan@example.com', '2025-10-27')
    # 2. 执行命令
    cursor.execute(sql_insert, params)
    # 3. 提交事务
    conn.commit() # 这是关键！没有它，数据不会被写入数据库
    print(f"成功插入 {cursor.rowcount} 行数据。")
except pyodbc.Error as e:
    print(f"插入数据失败: {e}")
    # 发生错误时回滚事务
    conn.rollback()
finally:
    if 'conn' in locals() and conn:
        cursor.close()
        conn.close()

示例：更新和删除

# 更新数据
try:
    cursor = conn.cursor()
    sql_update = "UPDATE Employees SET Email = ? WHERE Name = ?"
    cursor.execute(sql_update, 'new_zhangsan@example.com', '张三')
    conn.commit()
    print(f"成功更新 {cursor.rowcount} 行数据。")
except pyodbc.Error as e:
    print(f"更新数据失败: {e}")
    conn.rollback()
finally:
    # ... 关闭连接 ...
# 删除数据
try:
    cursor = conn.cursor()
    sql_delete = "DELETE FROM Employees WHERE Name = ?"
    cursor.execute(sql_delete, '张三')
    conn.commit()
    print(f"成功删除 {cursor.rowcount} 行数据。")
except pyodbc.Error as e:
    print(f"删除数据失败: {e}")
    conn.rollback()
finally:
    # ... 关闭连接 ...

高级用法

1 获取多行数据

使用 fetchall() 获取所有结果行，或 fetchmany(size) 获取指定数量的行。

cursor.execute("SELECT * FROM Employees")
# rows = cursor.fetchall() # 获取所有行，返回一个元组列表
rows = cursor.fetchmany(5) # 获取前5行
for row in rows:
    print(f"ID: {row.ID}, Name: {row.Name}, Email: {row.Email}")

2 使用上下文管理器 (with 语句)

pyodbc 的连接对象支持上下文管理器，可以自动处理连接的关闭,即使发生异常。

import pyodbc
# 连接字符串...
connection_string = '...'
try:
    # with 语句会自动在代码块结束后关闭连接
    with pyodbc.connect(connection_string) as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT TOP 3 * FROM Employees")
            for row in cursor.fetchall():
                print(row)
    # 连接在这里自动关闭
    print("连接已自动关闭。")
except pyodbc.Error as e:
    print(f"发生错误: {e}")

3 将查询结果转换为字典列表

默认情况下，pyodbc 的行对象是 pyodbc.Row 类型，可以通过索引访问，为了更方便地使用列名,可以将其转换为字典。

import pyodbc
connection_string = '...'
def query_to_dict_list(conn, sql_query, params=None):
    cursor = conn.cursor()
    if params:
        cursor.execute(sql_query, params)
    else:
        cursor.execute(sql_query)
    # 获取列名
    columns = [column[0] for column in cursor.description]
    # 将每一行转换为字典
    result = []
    for row in cursor.fetchall():
        result.append(dict(zip(columns, row)))
    return result
try:
    with pyodbc.connect(connection_string) as conn:
        sql = "SELECT * FROM Employees WHERE Department = ?"
        employees = query_to_dict_list(conn, sql, params='IT')
        for emp in employees:
            print(emp)
except pyodbc.Error as e:
    print(f"查询失败: {e}")

4 处理日期时间类型

SQL Server 的 datetime 类型与 Python 的 datetime 对象可以直接交互，pyodbc 会自动处理转换。

from datetime import datetime
try:
    with pyodbc.connect(connection_string) as conn:
        cursor = conn.cursor()
        # Python datetime 对象
        hire_date = datetime.now()
        # 插入 datetime
        sql = "INSERT INTO Employees (Name, HireDate) VALUES (?, ?)"
        cursor.execute(sql, '李四', hire_date)
        conn.commit()
        # 查询 datetime
        cursor.execute("SELECT HireDate FROM Employees WHERE Name = '李四'")
        db_date = cursor.fetchone()[0]
        print(f"从数据库读取的日期: {db_date}, 类型: {type(db_date)}")
except pyodbc.Error as e:
    print(f"操作失败: {e}")

替代方案：pymssql

除了 pyodbc，还有一个纯 Python 实现的库叫做 pymssql，它不需要配置 ODBC 驱动，安装和使用更简单，但性能通常不如 pyodbc。

安装 pymssql

pip install pymssql

使用 pymssql 示例

import pymssql
# 连接参数
server = 'your_server_name'
user = 'your_username'
password = 'your_password'
database = 'your_database_name'
try:
    # 建立连接
    conn = pymssql.connect(server, user, password, database)
    print("连接成功！")
    # 创建游标
    cursor = conn.cursor(as_dict=True) # as_dict=True 可以直接返回字典
    # 执行查询
    cursor.execute("SELECT * FROM Employees WHERE Department = %s", 'IT')
    # 获取结果
    for row in cursor:
        print(f"ID: {row['ID']}, Name: {row['Name']}")
except Exception as e:
    print(f"操作失败: {e}")
finally:
    if 'conn' in locals() and conn:
        cursor.close()
        conn.close()
        print("连接已关闭。")

总结与推荐

我强烈推荐使用 pyodbc，因为它更稳定、性能更好，是业界标准，虽然初次配置需要安装 ODBC 驱动,但这通常是一次性的工作。