☰
pymssql 连接 SQL Server 实战:安装、查询、事务避坑与性能优化
2026/10/11 14:01:34 网站建设 项目流程

简介:一份面向Python开发者与SQL Server使用者的入门教程PDF,系统讲解如何借助pymssql库完成微软数据库的连接与操作。内容覆盖连接数据库、游标使用注意事项、游标返回字典类型、with上下文管理器以及调用存储过程等核心环节,并配有可运行的代码示例,适合需要快速上手pymssql或梳理基础用法的读者。资源包仅包含1个PDF文件,压缩后大小约58KB,体积小巧,便于离线阅读与随时查阅。目前已有501人学习下载。通过文中对常见查询顺序冲突、参数传递、事务提交等细节的说明,读者能够避开典型踩坑点,掌握从建表、插入到查询、存储过程调用的完整流程,为后续在Python项目中集成SQL Server打下扎实基础。

1. pymssql 是什么:不折腾 ODBC 驱动就能连 SQL Server

某次临时取数,需求说十分钟后要一份订单明细,机器上只有 Python 3,我靠一条pip install pymssql就搞定,而同事那边为装 pyodbc 的 ODBC 驱动折腾了半小时。pymssql 是 Python 连接 Microsoft SQL Server 的老牌库,它不走系统 ODBC,而是内置 FreeTDS 直接和数据库服务器说 TDS 协议,跨平台体验相对省心。这篇文章把我平时用 pymssql 连 Mssql 的完整套路拆开:从安装、最小连接、查询、写入到最能劝退新手的编码和事务坑,照着跑就能把数据拉出来。适合场景很明确:临时脚本取数、批量搬运、跨平台部署,数据量在千万行以内都不需要换方案。

2. 装好 pymssql:为什么选它以及两个平台的安装差异

2.1 为什么选 pymssql:和 pyodbc 的差别

Python 连 SQL Server 常见方案就两个:pymssql 和 pyodbc。pyodbc 更通用,能连各种数据库,但它本质是 ODBC API 的封装,真正干活的是系统里的 ODBC Driver。SQL Server 的 ODBC Driver 在 Windows 上还好,到了 Linux 要装 unixODBC 和数据库官方驱动,还要维护 DSN 或连接字符串里的驱动名。报错经常是驱动名不对,一查是版本名对不上,这种排查最费时间。

pymssql 把 FreeTDS 直接包进来,TDS 协议层自己处理,连接参数就是一个 Python 字典,不涉及系统级配置。对于脚本型、临时型、中小规模数据搬运,pymssql 上手成本明显更低。如果你的项目里已经统一用 SQLAlchemy,两个库都能接,但小脚本单独取数,我一般直接上 pymssql,少一层中间件就少一个变量。

对比项pymssqlpyodbc
底层实现内置 FreeTDS,直连 TDS 协议调用系统 ODBC Driver
安装依赖Linux 需 freetds-dev,Windows 有预编译包需要额外装对应版本 ODBC Driver
连接配置全在 Python 参数里DSN 或驱动名,错一个字符就报错
适用场景脚本取数、批量读写、跨平台部署需要同时连多种数据库时

这个表不是要分高下,而是帮你快速判断。如果你的机器上已经装好了全套 ODBC 环境,pyodbc 也没问题;但如果是从零开始、只想快点把数据拿出来,pymssql 的路径最短。

2.2 Windows 与 Linux 的安装命令

Windows 下最省事,pip 直接拉预编译包,不需要额外装 FreeTDS。Linux 下如果找不到对应架构的预编译包,pip 会现场编译,这时候需要 FreeTDS 的开发头文件。

# Windows:pip 直接装预编译包 pip install pymssql # Ubuntu / Debian:先装 FreeTDS 开发包,再装 sudo apt-get update sudo apt-get install -y freetds-dev pip install pymssql # 如果 pip 安装时现场编译,还需要编译工具链 sudo apt-get install -y build-essential python3-dev

逻辑说明:Windows 的安装包是编译好的 wheel,FreeTDS 被打进去了,所以一条命令就够。Linux 下 freetds-dev 提供编译 pymssql C 扩展所需的头文件,build-essential 提供 gcc 等工具,python3-dev 提供 Python 头文件,缺哪个都会在编译阶段报错。

参数说明:如果内网环境没有外网源,常见做法是找一台能联网的机器把 wheel 包下载下来,传到内网用pip install 文件名.whl安装,同样能跑。装完后别急着写业务代码,先看下一小节的验证步骤。

2.3 装完后先验证 FreeTDS 与连通性

很多人装完直接写连接代码,连不上就开始怀疑人生。我的习惯是先做两步验证:第一步确认 pymssql 本身可用,第二步确认网络端口通不通。

import pymssql # 确认模块版本能正常导入 print("pymssql 版本:", pymssql.__version__)
import socket # 测 TCP 连通性:IP 换成你的数据库服务器地址 s = socket.create_connection(("192.168.1.10", 1433), timeout=5) print("1433 端口可连通") s.close()

逻辑说明:第一段代码验证 Python 侧导入没问题,第二段代码验证数据库服务器 1433 端口能建立 TCP 连接。端口通说明网络和防火墙基本没问题,但协议是否匹配要等真正 connect 才知道;如果 1433 不通,优先排查 SQL Server 是否开启了 TCP/IP 协议,这一步是新手第一次连不上的首因。

提示:别只看安装成功就万事大吉。pymssql 在编译时会把 FreeTDS 的行为编进去,TDS 版本过低会影响高版本 SQL Server 的登录,这个在第 5 章会展开讲。

3. 第一次连接与查询:最小代码和连接参数逐项拆解

3.1 最小连接代码:先跑通再说

这一节的目标是让你在五分钟内看到第一条查询结果。最小连接只需要四个参数:服务地址、账号、密码、数据库名。

import pymssql # 最小连接参数:server 可以是 IP 或主机名 conn = pymssql.connect( server="127.0.0.1", user="sa", password="YourStrongPass", database="demo_db", charset="utf-8", ) cur = conn.cursor() cur.execute("SELECT @@VERSION AS version_info") row = cur.fetchone() print(row[0]) conn.close()

逻辑说明:这段代码做两件事,建立连接和验证数据库版本。SELECT @@VERSION是 SQL Server 的系统查询,不涉及业务表,用来确认账号能登录、当前连接的数据库实例没问题。charset="utf-8"一开始就设好,能少踩很多中文乱码的坑。

参数说明:conn.close()放在最后,但实际项目里如果中间报错,连接会泄漏。更稳的写法是用with语句,第 6 章会专门说。现在先跑通这条最小链路,再往下看连接参数。

3.2 连接参数逐项拆解:一份可以直接抄的参数表

pymssql 的connect()参数不算多,但每个都可能成为坑。我按优先级列一份常用参数表,照着填基本不会错。

参数作用常见值 / 说明
server数据库地址IP 或主机名;命名实例写成host\instance
port端口默认 1433,非默认端口时单独传
user登录账号SQL Server 身份验证的账号
password账号密码建议从环境变量读取
database初始数据库不传也能连,但每条 SQL 都要写全库名
charset字符编码中文环境优先utf-8
timeout查询超时秒数0 表示不限制,生产脚本建议设个值
login_timeout登录超时秒数默认 15,内网慢环境可以调到 30
tds_versionTDS 协议版本日常不填,自动协商;连不上时再显式指定
autocommit是否自动提交默认 False,写操作需要手动 commit
appname应用名可选,方便数据库管理员定位你的连接

参数说明:user和password是 SQL Server 身份验证的账号。如果数据库只开了 Windows 身份验证模式,用账号密码登录会被拒绝,需要找管理员改配置或换连接方式,这是常见盲区。database建议传,不传的话每条查询都得写库名.dbo.表名,代码冗长不说,还容易把库弄错。

注意:pymssql 处理 Windows 身份验证不太方便。如果环境只允许 Windows 认证,常见做法是换 pyodbc 用Trusted_Connection=Yes,或者让数据库管理员给你开一个 SQL 登录账号,别在 pymssql 上死磕。

3.3 把查询结果读出来:fetchall、fetchone 与游标遍历

连接建立之后,查询结果有三种读法。行为一致,区别在内存占用和代码风格。

cur.execute("SELECT id, name, age FROM dbo.users ORDER BY id") # 方式一:一次性取全部,适合数据量小 rows = cur.fetchall() for row in rows: print(row[0], row[1], row[2]) # 方式二:循环 fetchone,省内存,适合大结果集 row = cur.fetchone() while row: print(row) row = cur.fetchone() # 方式三:游标本身可迭代,写法最简洁 for row in cur: print(row)

逻辑说明:fetchall()会把所有结果放进内存,几十万行没问题,几百万行开始有压力,如果结果集很大,循环fetchone()是稳妥选择。游标本身的迭代本质上也是逐行取,但代码更短。三种方式的返回结构一样,每行是一个元组,按下标访问列。

这里有个新手容易踩的细节:游标指针是移动的。如果先fetchone()取了一行,再fetchall(),拿到的是从第二行开始的所有剩余数据,不是全部。这个行为不是 pymssql 特有,所有数据库游标都这样,但初次接触时容易懵。

3.4 列名访问与字段类型映射:拿到数据后长什么样

默认每行是元组,下标访问在字段多的时候容易写错。pymssql 支持as_dict=True让每行变成字典,代码可读性高很多。

cur = conn.cursor(as_dict=True) cur.execute("SELECT id, name, age FROM dbo.users") for row in cur: print(row["name"], row["age"])

逻辑说明:as_dict=True的开销比元组略大,但脚本场景下无所谓。字段重名时字典会丢一个,查询里最好给重复列名起别名。

类型映射是另一个值得提前知道的点。pymssql 把 SQL Server 字段转成 Python 类型时有自己的规则,提前知道能少踩精度坑。

SQL Server 类型pymssql 返回类型备注
int / bigintint直接对应
varchar / nvarcharstr前提是 charset 设置正确
decimal / numericfloat 或 Decimal受 TDS 转换影响,精度要小心
datetime / datetime2datetime.datetime可直接比较和格式化
bitboolTrue / False
uniqueidentifierstr形如a1b2c3d4-...

decimal这个类型在第 5 章会专门展开,金额类字段一定要看那一节。

4. 写入与事务:把 execute、commit、executemany 用明白

4.1 参数化查询:别用 f-string 拼 SQL

查数据总有带条件的时候。很多新人习惯用 f-string 直接拼 SQL,字段值是数字还好,遇到字符串就麻烦了:引号要转义、特殊字符会报错,更危险的是 SQL 注入。pymssql 的参数化写法很简单。

cur = conn.cursor() # %s 是字符串占位符,值作为第二个参数传入 cur.execute( "SELECT * FROM dbo.users WHERE name = %s", ("张三",), ) # %d 是 pymssql 的整型占位符,不是 Python 格式化 cur.execute( "SELECT * FROM dbo.users WHERE id = %d", (42,), )

逻辑说明:第一段是字符串条件的标准写法,第二段是整型条件。注意%d是 pymssql 自带的扩展语法,不是 Python 字符串格式化里的%d,它只接受整型值,传字符串进去会报错或查不到数据。

参数说明:参数个数要和占位符一一对应,多一个少一个都会报参数数量不匹配。如果条件特别多,用列表或元组传参都行,但顺序不能乱。别用 f-string 拼 SQL,等踩到引号转义的坑再回来改,浪费时间。这里不是危言耸听,很多生产事故就是拼 SQL 拼出来的。

4.2 事务边界:commit、rollback 与 autocommit

pymssql 默认不自动提交事务。这意味着execute()写完数据后,如果不调用commit(),连接关闭时事务会回滚,数据等于没写。这个行为坑过很多人:脚本跑完显示成功,查库发现啥也没有。

conn = pymssql.connect( server="...", user="...", password="...", database="...", charset="utf-8", ) cur = conn.cursor() cur.execute( "INSERT INTO dbo.users (name, age) VALUES (%s, %d)", ("王五", 25), ) # 确认数据没问题再提交,数据才真正落库 conn.commit() # 如果发现写错了,在 commit 之前调用 rollback 可以反悔 conn.rollback()

逻辑说明:默认autocommit=False时,execute()只是把语句发给服务器,事务在会话内挂着。commit()之后才生效,rollback()是数据库后悔药,前提是在 commit 之前调用。如果连接直接关闭,未提交的事务一样会回滚。

# 临时脚本或初始化数据时,可以开自动提交 conn.autocommit(True)

参数说明:autocommit(True)让每条语句执行后立即生效,适合临时清表、初始化数据这种场景。生产脚本我坚持手动 commit,理由是一次事务里可能涉及多张表,要么全部成功要么全部回滚,自动提交会把事务拆碎,出问题没法收场。

注意:commit 只是结束当前事务,不是重连。事务中出现错误后,建议进入异常分支主动 rollback,别假装没发生继续往下写。

4.3 executemany 批量写入:一次传一批,别逐条循环

批量插入是 pymssql 最常用的功能之一。逐条execute()循环能跑,但每一条都是一次网络往返,几万行数据会慢到怀疑人生。executemany()把一批参数一次性发给服务器,效率高一个量级。

data = [("A001", 100), ("A002", 200), ("A003", 300)] cur = conn.cursor() cur.executemany( "INSERT INTO dbo.orders (order_no, amount) VALUES (%s, %d)", data, ) conn.commit()

逻辑说明:executemany()接收两个参数:SQL 模板和参数列表。它内部按批发送,不是逐条独立往返。注意它不会自动提交,批量写完后仍然需要手动commit()。

数据量大的时候,单次executemany()塞几万行会让事务日志和锁的压力变大,别人的查询会被阻塞。我一般会手动分批,每批 500 到 1000 行,一批提交一次。

batch_size = 500 for i in range(0, len(data), batch_size): sub = data[i:i + batch_size] cur.executemany(sql, sub) conn.commit()

参数说明:batch_size不是一个固定值。取决于单行字段的宽度和网络延迟:字段多、有大文本或二进制数据时调小到 200 左右;表结构简单、网络稳定时提到 1000 也能接受。判断标准很简单:跑一次看数据库 CPU 和阻塞情况,有明显阻塞就调小。

5. pymssql 避坑:连接失败、中文乱码、慢查询的 5 个现场

5.1 连接超时:TDS 协议版本太低

现象:connect()卡十几秒后报TimeoutError,或者登录时报DB-Lib error message 20002, severity 9: Adaptive Server connection failed这类错误。

原因:FreeTDS 和 SQL Server 协商 TDS 协议版本时没谈拢。老版本编译的 FreeTDS 默认协议版本偏低,高版本 SQL Server 的登录流程对不上;另一种可能是端口不是默认的 1433。

解决:连接参数里显式指定tds_version,并调大登录超时时间。

conn = pymssql.connect( server="...", user="...", password="...", login_timeout=30, tds_version="7.3", )

逻辑说明:7.3是 SQL Server 2008 及以后常用的 TDS 版本号,遇到连接失败时值得一试。login_timeout=30给慢网络留出余量,避免默认 15 秒不够用。

诊断顺序也很重要:先确认端口通不通(第 2 章的 socket 测试),再查 SQL Server 是否开了 TCP/IP 协议,最后才怀疑协议版本。这个顺序能省一半的排查时间。

5.2 中文乱码:charset 与终端编码的三角问题

现象:从数据库查出来的中文变成?、ï之类;写入数据库后中文变成问号。

原因:中文乱码经常不是一层问题。第一层是 pymssql 连接的charset没设置,默认编码和数据库不匹配;第二层是 Windows 终端 print 时控制台用 GBK,Python 输出 UTF-8,显示就乱了;第三层是写入时 SQL 里的字符串字面量没加N前缀,nvarchar 字段存不进中文。

解决:连接参数统一charset="utf-8";验证乱码是多层问题还是单层问题;写入时 SQL 字符串加N前缀。

with pymssql.connect(..., charset="utf-8") as conn: cur = conn.cursor() cur.execute("SELECT N'中文测试' AS text_val") print(cur.fetchone()[0])

逻辑说明:N前缀是 SQL Server 里 nvarchar 字面量的标志,告诉服务器这个字符串按 Unicode 处理。如果只用普通字符串字面量,字段类型是 varchar 时可能正常,碰到 nvarchar 就出问题。

注意:如果 Python 打印出来乱码,但把数据写到文件里是正常的,那是终端编码问题,不是 pymssql 的问题,别在连接参数上浪费时间。

5.3 数字精度:decimal 变成 float 之后对不上账

现象:金额字段查询出来变成100.1而不是100.10,累计求和时出现0.1 + 0.2 = 0.30000000000000004这种经典误差。

原因:pymssql 底层把decimal/numeric转成 Python 类型时,常见结果是float。float是二进制浮点,无法精确表示十进制小数,累加运算会出现舍入误差。只展示两三位小数看不出问题,一求和就露馅。

解决:如果需要精确计算,让数据库在 SQL 阶段转成字符串,Python 再用Decimal处理。

from decimal import Decimal cur.execute("SELECT CAST(amount AS VARCHAR(20)) FROM dbo.orders") amount_str = cur.fetchone()[0] amount = Decimal(amount_str)

逻辑说明:CAST(amount AS VARCHAR(20))让 SQL Server 先把数值转成字符串,避开浮点转换。Python 的Decimal处理字符串构造的十进制数是精确的。代价是数据库端多做一次转换,但对金额对账这种场景值得。

如果只是展示,round(value, 2)也能应付;但涉及累加、对账、报表汇总,必须走Decimal这条路。

5.4 查询慢:先在数据库端定位,再让 Python 背锅

现象:同样的 SQL 在数据库管理工具里秒回,放到 Python 里跑几十秒;或者整个库里某个查询长期占用 CPU。

原因:常见有四种。一是参数嗅探导致执行计划偏离,二是 Python 端循环里逐条查库形成 N+1 查询,三是查询条件字段没有索引,四是结果集太大客户端 fetch 本身慢。

解决:先在数据库端确认执行时间,再优化 Python 侧。

SET STATISTICS TIME ON; SELECT * FROM dbo.orders WHERE status = 'PENDING';

逻辑说明:SET STATISTICS TIME ON会显示语句的编译时间和执行时间。如果数据库端执行很快,问题在 Python 侧;如果数据库端执行就慢,应该先优化 SQL 本身,而不是改 Python 代码。

Python 侧的优化原则是减少数据量:只 SELECT 需要的列、加 WHERE 条件、用分页。

cur.execute( "SELECT order_no, amount FROM dbo.orders " "WHERE status = %s " "ORDER BY id OFFSET %d ROWS FETCH NEXT 1000 ROWS ONLY", ("PENDING", offset), )

逻辑说明:OFFSET ... FETCH NEXT是 SQL Server 的分页语法,一次只取 1000 行,避免一次性拉回全表。offset是页码偏移,翻页时重新计算。

5.5 多线程共享连接:报错飘忽不定,不是玄学是线程安全

现象:多线程脚本里共享同一个连接,有时报错有时丢数据,错误信息还不固定,排查起来像玄学。

原因:pymssql 的Connection和Cursor底层是 C 扩展的 FreeTDS 句柄,不是线程安全的。多线程同时对同一个连接执行查询,内部状态互相踩踏,行为不可预测。这不是概率问题,是必然问题,只是触发时机随机。

解决:每个线程单独建连接,或者引入连接池。推荐后者,连接开销不会成倍增长。

from dbutils.pooled_db import PooledDB import pymssql pool = PooledDB( creator=pymssql, maxconnections=10, server="...", user="...", password="...", database="...", charset="utf-8", ) conn = pool.connection() cur = conn.cursor() cur.execute("SELECT * FROM dbo.users") print(cur.fetchone()) conn.close() # 归还连接给池子,不是真正断开

逻辑说明:PooledDB的creator传入 pymssql 这个类,连接参数直接映射到pymssql.connect()。maxconnections=10控制池子上限,避免连接数打满数据库。线程从池子里拿连接,用完归还,互不干扰。

注意:即使只是读操作,也别赌共享连接没问题。线程安全问题在低并发时可能几个月不触发,一旦触发就是线上事故。

6. 收尾进阶:上下文管理器与一个连接验证习惯

6.1 用 with 管理连接与游标

手写conn.close()在异常时会漏执行,用with语句是最稳的写法。

import pymssql with pymssql.connect( server="...", user="...", password="...", database="...", charset="utf-8", ) as conn: with conn.cursor() as cur: cur.execute("SELECT @@VERSION") print(cur.fetchone()[0])

逻辑说明:with conn在退出代码块时自动关闭连接,with cur自动关闭游标。但要记住,with conn不会自动提交事务,写操作仍需要手动commit()。

6.2 连接前的一个验证习惯:参数从环境变量读

硬编码数据库密码是翻车根源之一。我见过把测试库密码顺手粘到生产脚本里的,全量更新跑完才发现连错库。后来的习惯是所有连接参数从环境变量读取,不设默认值。

import os import pymssql config = { "server": os.environ.get("MSSQL_HOST", "127.0.0.1"), "user": os.environ.get("MSSQL_USER", "sa"), "password": os.environ.get("MSSQL_PASS"), "database": os.environ.get("MSSQL_DB", "demo_db"), "charset": "utf-8", } def ping_db(cfg): with pymssql.connect(**cfg) as conn: with conn.cursor() as cur: cur.execute("SELECT @@VERSION") return cur.fetchone()[0]

逻辑说明:os.environ.get从环境变量读参数,密码字段没有默认值,缺了就报错而不是用错误密码连。ping_db函数统一做连通性检查,所有脚本接数据库之前先调它,确认协议、账号、网络都没问题再碰业务表。

6.3 一个收尾的教训

我入行时踩过最贵的一坑,是把测试环境的连接串复制到生产脚本里,跑完全量更新才发现连错了库。从那以后,server 参数一律从配置读取,禁止在代码里写默认数据库地址,环境变量缺了就立刻报错。这套习惯帮我挡掉了不少潜在事故,也希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询