简介一份面向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-devWindows 有预编译包需要额外装对应版本 ODBC Driver连接配置全在 Python 参数里DSN 或驱动名错一个字符就报错适用场景脚本取数、批量读写、跨平台部署需要同时连多种数据库时这个表不是要分高下而是帮你快速判断。如果你的机器上已经装好了全套 ODBC 环境pyodbc 也没问题但如果是从零开始、只想快点把数据拿出来pymssql 的路径最短。2.2 Windows 与 Linux 的安装命令Windows 下最省事pip 直接拉预编译包不需要额外装 FreeTDS。Linux 下如果找不到对应架构的预编译包pip 会现场编译这时候需要 FreeTDS 的开发头文件。# Windowspip 直接装预编译包 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 的安装包是编译好的 wheelFreeTDS 被打进去了所以一条命令就够。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), timeout5) 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( server127.0.0.1, usersa, passwordYourStrongPass, databasedemo_db, charsetutf-8, ) cur conn.cursor() cur.execute(SELECT VERSION AS version_info) row cur.fetchone() print(row[0]) conn.close()逻辑说明这段代码做两件事建立连接和验证数据库版本。SELECT VERSION是 SQL Server 的系统查询不涉及业务表用来确认账号能登录、当前连接的数据库实例没问题。charsetutf-8一开始就设好能少踩很多中文乱码的坑。参数说明conn.close()放在最后但实际项目里如果中间报错连接会泄漏。更稳的写法是用with语句第 6 章会专门说。现在先跑通这条最小链路再往下看连接参数。3.2 连接参数逐项拆解一份可以直接抄的参数表pymssql 的connect()参数不算多但每个都可能成为坑。我按优先级列一份常用参数表照着填基本不会错。参数作用常见值 / 说明server数据库地址IP 或主机名命名实例写成host\instanceport端口默认 1433非默认端口时单独传user登录账号SQL Server 身份验证的账号password账号密码建议从环境变量读取database初始数据库不传也能连但每条 SQL 都要写全库名charset字符编码中文环境优先utf-8timeout查询超时秒数0 表示不限制生产脚本建议设个值login_timeout登录超时秒数默认 15内网慢环境可以调到 30tds_versionTDS 协议版本日常不填自动协商连不上时再显式指定autocommit是否自动提交默认 False写操作需要手动 commitappname应用名可选方便数据库管理员定位你的连接参数说明user和password是 SQL Server 身份验证的账号。如果数据库只开了 Windows 身份验证模式用账号密码登录会被拒绝需要找管理员改配置或换连接方式这是常见盲区。database建议传不传的话每条查询都得写库名.dbo.表名代码冗长不说还容易把库弄错。注意pymssql 处理 Windows 身份验证不太方便。如果环境只允许 Windows 认证常见做法是换 pyodbc 用Trusted_ConnectionYes或者让数据库管理员给你开一个 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_dictTrue让每行变成字典代码可读性高很多。cur conn.cursor(as_dictTrue) cur.execute(SELECT id, name, age FROM dbo.users) for row in cur: print(row[name], row[age])逻辑说明as_dictTrue的开销比元组略大但脚本场景下无所谓。字段重名时字典会丢一个查询里最好给重复列名起别名。类型映射是另一个值得提前知道的点。pymssql 把 SQL Server 字段转成 Python 类型时有自己的规则提前知道能少踩精度坑。SQL Server 类型pymssql 返回类型备注int / bigintint直接对应varchar / nvarcharstr前提是 charset 设置正确decimal / numericfloat 或 Decimal受 TDS 转换影响精度要小心datetime / datetime2datetime.datetime可直接比较和格式化bitboolTrue / Falseuniqueidentifierstr形如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 与 autocommitpymssql 默认不自动提交事务。这意味着execute()写完数据后如果不调用commit()连接关闭时事务会回滚数据等于没写。这个行为坑过很多人脚本跑完显示成功查库发现啥也没有。conn pymssql.connect( server..., user..., password..., database..., charsetutf-8, ) cur conn.cursor() cur.execute( INSERT INTO dbo.users (name, age) VALUES (%s, %d), (王五, 25), ) # 确认数据没问题再提交数据才真正落库 conn.commit() # 如果发现写错了在 commit 之前调用 rollback 可以反悔 conn.rollback()逻辑说明默认autocommitFalse时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_timeout30, tds_version7.3, )逻辑说明7.3是 SQL Server 2008 及以后常用的 TDS 版本号遇到连接失败时值得一试。login_timeout30给慢网络留出余量避免默认 15 秒不够用。诊断顺序也很重要先确认端口通不通第 2 章的 socket 测试再查 SQL Server 是否开了 TCP/IP 协议最后才怀疑协议版本。这个顺序能省一半的排查时间。5.2 中文乱码charset 与终端编码的三角问题现象从数据库查出来的中文变成?、ï之类写入数据库后中文变成问号。原因中文乱码经常不是一层问题。第一层是 pymssql 连接的charset没设置默认编码和数据库不匹配第二层是 Windows 终端 print 时控制台用 GBKPython 输出 UTF-8显示就乱了第三层是写入时 SQL 里的字符串字面量没加N前缀nvarchar 字段存不进中文。解决连接参数统一charsetutf-8验证乱码是多层问题还是单层问题写入时 SQL 字符串加N前缀。with pymssql.connect(..., charsetutf-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 端循环里逐条查库形成 N1 查询三是查询条件字段没有索引四是结果集太大客户端 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( creatorpymssql, maxconnections10, server..., user..., password..., database..., charsetutf-8, ) conn pool.connection() cur conn.cursor() cur.execute(SELECT * FROM dbo.users) print(cur.fetchone()) conn.close() # 归还连接给池子不是真正断开逻辑说明PooledDB的creator传入 pymssql 这个类连接参数直接映射到pymssql.connect()。maxconnections10控制池子上限避免连接数打满数据库。线程从池子里拿连接用完归还互不干扰。注意即使只是读操作也别赌共享连接没问题。线程安全问题在低并发时可能几个月不触发一旦触发就是线上事故。6. 收尾进阶上下文管理器与一个连接验证习惯6.1 用 with 管理连接与游标手写conn.close()在异常时会漏执行用with语句是最稳的写法。import pymssql with pymssql.connect( server..., user..., password..., database..., charsetutf-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 参数一律从配置读取禁止在代码里写默认数据库地址环境变量缺了就立刻报错。这套习惯帮我挡掉了不少潜在事故也希望帮到你。本文还有配套的精品资源点击获取