pymssql 是一个纯 Python 实现的 MSSQL 驱动程序，它基于 FreeTDS 库，它简单易用，是 Python 连接 SQL Server 的常用选择之一。

安装 pymssql

你需要安装这个库，最简单的方式是使用 pip。

pip install pymssql

注意: 在某些系统（特别是 Windows）上，你可能需要先安装 FreeTDS 的开发库，但通常 pip 会自动处理依赖关系，如果遇到问题,可能需要根据你的操作系统进行额外配置。

基本连接

使用 pymssql 连接到 SQL Server 数据库非常直接，核心是 pymssql.connect() 函数。

连接参数

pymssql.connect() 接收以下主要参数：

• server: SQL Server 的主机名或 IP 地址。
• user: 登录用户名。
• password: 登录密码。
• database: 要连接的默认数据库名称。
• port: SQL Server 的端口号，默认是 1433。
• charset: 字符集，通常使用 'utf-8' 以避免中文乱码。
• as_dict: 一个非常重要的参数，如果设置为 True，查询结果会以字典列表的形式返回，列名作为键，如果为 False（默认），结果会是元组列表。强烈推荐设置为 True,因为它让代码更具可读性。

示例：建立连接

import pymssql
# --- 连接信息 ---
# 建议将敏感信息（如密码）存储在环境变量或配置文件中，而不是硬编码在代码里。
server = "your_server_name"
user = "your_username"
password = "your_password"
database = "your_database_name"
try:
    # 建立连接
    conn = pymssql.connect(
        server=server,
        user=user,
        password=password,
        database=database,
        charset='utf-8',
        as_dict=True  # 推荐使用字典形式获取结果
    )
    print("连接成功！")
    # ... 在这里执行数据库操作 ...
except Exception as e:
    print(f"连接数据库失败: {e}")
finally:
    # 确保连接被关闭
    if 'conn' in locals() and conn:
        conn.close()
        print("连接已关闭。")

执行 SQL 语句

一旦连接建立，你需要一个游标（cursor）来执行 SQL 语句。

1 执行查询 (SELECT)

使用 cursor.execute() 执行查询，然后用 cursor.fetchall(), cursor.fetchone() 或 cursor.fetchmany() 获取结果。

• fetchall(): 获取所有剩余的行。
• fetchone(): 获取下一行。
• fetchmany(size): 获取指定数量的行。

import pymssql
# (使用上面相同的连接信息)
conn = pymssql.connect(server="your_server", user="your_user", password="your_password", database="your_db", as_dict=True)
try:
    with conn.cursor() as cursor:  # 使用 with 语句可以确保游标被正确关闭
        # 1. 执行查询
        cursor.execute("SELECT TOP 10 * FROM YourTableName")
        # 2. 获取所有结果
        # 因为 as_dict=True，result 是一个字典列表
        result = cursor.fetchall()
        print(f"查询到 {len(result)} 条记录:")
        for row in result:
            print(row) # 打印每一行字典
            # 访问特定列: print(row['YourColumnName'])
        # --- 另一个例子：使用 fetchone ---
        cursor.execute("SELECT COUNT(*) AS total_count FROM YourTableName")
        row = cursor.fetchone() # 获取第一行（也是唯一一行）
        if row:
            print(f"表 'YourTableName' 总共有 {row['total_count']} 条记录.")
except Exception as e:
    print(f"查询出错: {e}")
    # 发生错误时可以回滚事务
    conn.rollback()
finally:
    conn.close()

2 执行非查询语句 (INSERT, UPDATE, DELETE)

对于修改数据的操作，执行完毕后必须调用 conn.commit() 来提交事务,否则更改不会永久保存到数据库。

import pymssql
# (使用上面相同的连接信息)
conn = pymssql.connect(server="your_server", user="your_user", password="your_password", database="your_db")
try:
    with conn.cursor() as cursor:
        # --- 插入数据 ---
        # 使用参数化查询来防止 SQL 注入攻击！
        insert_sql = "INSERT INTO YourTableName (Name, Age, City) VALUES (%s, %s, %s)"
        new_data = ("张三", 30, "北京")
        cursor.execute(insert_sql, new_data)
        # --- 更新数据 ---
        update_sql = "UPDATE YourTableName SET Age = %s WHERE Name = %s"
        cursor.execute(update_sql, (31, "张三"))
        # --- 删除数据 ---
        # delete_sql = "DELETE FROM YourTableName WHERE Name = %s"
        # cursor.execute(delete_sql, ("张三",))
        # 提交事务
        conn.commit()
        print("操作成功，已提交事务。")
except Exception as e:
    print(f"操作出错: {e}")
    # 发生错误时回滚事务
    conn.rollback()
finally:
    conn.close()

⚠️ 重要安全提示： 永远不要使用 Python 的字符串格式化（如 f"query WHERE name = '{user_input}'"）来构建 SQL 查询。 这极易导致 SQL 注入攻击。始终使用参数化查询，如 cursor.execute("query WHERE name = %s", (user_input,))。pymssql 会安全地处理参数的转义。

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

为了确保连接和游标在使用完毕后被正确关闭（即使在发生异常时），强烈推荐使用 with 语句。

• with conn.cursor() as cursor:: 自动管理游标的生命周期。
• with pymssql.connect(...) as conn:: 自动管理连接的生命周期。

import pymssql
# with 语句会自动处理连接的关闭
with pymssql.connect(server="your_server", user="your_user", password="your_password", database="your_db", as_dict=True) as conn:
    # with 语句会自动处理游标的关闭
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM YourTableName")
        for row in cursor.fetchall():
            print(row)
    # 离开 with conn.cursor() 块后，游标自动关闭
# 离开 with pymssql.connect(...) 块后，连接自动关闭

存储过程

pymssql 也支持调用存储过程。

import pymssql
conn = pymssql.connect(server="your_server", user="your_user", password="your_password", database="your_db")
try:
    with conn.cursor() as cursor:
        # 假设有一个名为 'GetEmployee' 的存储过程，接受一个 @EmployeeID 参数
        # 并返回一个结果集
        cursor.callproc('GetEmployee', (101,))  # 参数必须是一个元组或列表
        # 获取存储过程返回的第一个结果集
        for row in cursor.fetchall():
            print(row)
        # 如果存储过程有多个结果集，需要使用 nextset() 来获取下一个
        # cursor.nextset()
except Exception as e:
    print(f"调用存储过程出错: {e}")
    conn.rollback()
finally:
    conn.close()

完整示例代码

下面是一个整合了连接、查询、插入、提交和异常处理的完整示例。

import pymssql
import time
# --- 配置信息 ---
# 在实际应用中，这些应该从配置文件或环境变量中读取
DB_CONFIG = {
    "server": "localhost",  # 或你的服务器IP
    "user": "sa",
    "password": "your_strong_password",
    "database": "TestDB",
    "charset": "utf-8",
    "as_dict": True
}
def create_sample_table_if_not_exists(conn):
    """如果表不存在，则创建一个示例表"""
    with conn.cursor() as cursor:
        cursor.execute("""
        IF OBJECT_ID('Employees', 'U') IS NULL
        CREATE TABLE Employees (
            id INT PRIMARY KEY IDENTITY(1,1),
            name NVARCHAR(50) NOT NULL,
            department NVARCHAR(50),
            hire_date DATETIME
        );
        """)
        print("确保 'Employees' 表存在。")
def add_employee(conn, name, department):
    """添加一名新员工"""
    with conn.cursor() as cursor:
        sql = "INSERT INTO Employees (name, department, hire_date) VALUES (%s, %s, %s)"
        cursor.execute(sql, (name, department, time.strftime('%Y-%m-%d %H:%M:%S')))
        conn.commit()
        print(f"成功添加员工: {name}")
def get_all_employees(conn):
    """获取所有员工信息"""
    with conn.cursor() as cursor:
        cursor.execute("SELECT id, name, department, hire_date FROM Employees")
        employees = cursor.fetchall()
        print("\n当前员工列表:")
        if not employees:
            print("  (暂无员工)")
        for emp in employees:
            print(f"  ID: {emp['id']}, 姓名: {emp['name']}, 部门: {emp['department']}, 入职日期: {emp['hire_date']}")
# --- 主程序 ---
def main():
    conn = None
    try:
        # 1. 连接数据库
        conn = pymssql.connect(**DB_CONFIG)
        print("数据库连接成功。")
        # 2. 准备数据表
        create_sample_table_if_not_exists(conn)
        # 3. 查询并打印当前数据
        get_all_employees(conn)
        # 4. 添加新数据
        add_employee(conn, "李四", "技术部")
        add_employee(conn, "王五", "市场部")
        # 5. 再次查询，验证数据已添加
        get_all_employees(conn)
    except pymssql.OperationalError as e:
        print(f"数据库操作错误: {e}")
    except Exception as e:
        print(f"发生未知错误: {e}")
    finally:
        if conn:
            conn.close()
            print("\n数据库连接已关闭。")
if __name__ == "__main__":
    main()

• 安装: pip install pymssql
• 连接: 使用 pymssql.connect()，提供服务器、用户、密码和数据库名。
• 游标: 通过 conn.cursor() 获取游标来执行 SQL。
• 查询: cursor.execute() -> cursor.fetchall()/fetchone()。
• 增删改: cursor.execute() -> conn.commit()。
• 安全: 务必使用参数化查询 (%s 占位符) 来防止 SQL 注入。
• 资源管理: 强烈推荐使用 with 语句来自动管理连接和游标的生命周期,确保它们被正确关闭。
• 结果形式: 设置 as_dict=True 可以让结果以字典形式返回,提高代码可读性。