1. 环境准备与基础配置在开始操作SQL Server之前我们需要先准备好Python环境和必要的工具库。这里我推荐使用pymssql这个库它是Python连接SQL Server最常用的驱动之一。安装起来非常简单只需要一条命令pip install pymssql安装完成后我们还需要确保SQL Server已经正确配置。这里有个小坑我踩过好几次SQL Server默认不开启TCP/IP协议这会导致Python无法连接。解决方法是在SQL Server配置管理器中启用TCP/IP协议打开SQL Server配置管理器找到SQL Server网络配置选择你的实例通常是MSSQLSERVER右键启用TCP/IP协议重启SQL Server服务验证连接是否成功可以用Navicat等工具测试连接。如果Navicat能连上Python程序一般也能正常连接。2. 数据库与表的创建2.1 创建数据库我们先从创建数据库开始。这里有个技巧设置autocommitTrue可以避免创建数据库时的事务问题。来看具体代码import pymssql # 连接到服务器不指定具体数据库 conn pymssql.connect(serverlocalhost, usersa, passwordyour_password) # 创建游标设置自动提交 cursor conn.cursor(as_dictTrue) conn.autocommit(True) # 关键设置 # 检查现有数据库 cursor.execute(SELECT name FROM sys.databases) print(现有数据库) for db in cursor.fetchall(): print(db[name]) # 创建新数据库 create_db_sql IF NOT EXISTS(SELECT * FROM sys.databases WHERE name TestDB) CREATE DATABASE TestDB cursor.execute(create_db_sql) # 再次查看数据库列表 cursor.execute(SELECT name FROM sys.databases) print(\n创建后的数据库) for db in cursor.fetchall(): print(db[name]) conn.close()这段代码会先列出所有现有数据库然后创建TestDB数据库最后再次列出数据库确认创建成功。注意我们使用了sys.databases系统视图来查询数据库信息这是SQL Server的标准做法。2.2 创建数据表有了数据库后我们来创建一个简单的Persons表conn pymssql.connect(serverlocalhost, usersa, passwordyour_password, databaseTestDB) cursor conn.cursor() # 查看现有表 cursor.execute(SELECT name FROM sys.tables) print(现有表, cursor.fetchall()) # 创建Persons表 create_table_sql CREATE TABLE Persons ( ID INT PRIMARY KEY IDENTITY(1,1), Name NVARCHAR(50) NOT NULL, Age INT, CreateTime DATETIME DEFAULT GETDATE() ) cursor.execute(create_table_sql) conn.commit() # 注意这里需要显式提交 # 再次查看表 cursor.execute(SELECT name FROM sys.tables) print(创建后的表, cursor.fetchall()) conn.close()这里有几个注意点使用了IDENTITY(1,1)设置自增主键为Name字段指定了NVARCHAR类型以支持中文添加了CreateTime字段并设置默认值为当前时间创建表后需要调用commit()提交事务3. CRUD操作实战3.1 插入数据Create插入数据有两种方式单条插入和批量插入。批量插入效率更高特别是在需要插入大量数据时conn pymssql.connect(serverlocalhost, usersa, passwordyour_password, databaseTestDB) cursor conn.cursor(as_dictTrue) # 单条插入 cursor.execute(INSERT INTO Persons (Name, Age) VALUES (张三, 20)) # 批量插入 data [(李四, 25), (王五, 30), (赵六, 35)] cursor.executemany(INSERT INTO Persons (Name, Age) VALUES (%s, %s), data) conn.commit() # 查询验证 cursor.execute(SELECT * FROM Persons) for row in cursor.fetchall(): print(fID: {row[ID]}, 姓名: {row[Name]}, 年龄: {row[Age]}, 创建时间: {row[CreateTime]}) conn.close()3.2 查询数据Retrieve查询数据时pymssql提供了几种获取结果的方式conn pymssql.connect(serverlocalhost, usersa, passwordyour_password, databaseTestDB) cursor conn.cursor(as_dictTrue) # 获取全部结果 cursor.execute(SELECT * FROM Persons) print(全部记录) all_records cursor.fetchall() for row in all_records: print(row) # 获取部分结果 cursor.execute(SELECT * FROM Persons) print(\n前2条记录) some_records cursor.fetchmany(2) for row in some_records: print(row) # 逐行获取 cursor.execute(SELECT * FROM Persons) print(\n逐行获取) while True: row cursor.fetchone() if row is None: break print(row) conn.close()3.3 更新数据Update更新操作也支持单条和批量模式conn pymssql.connect(serverlocalhost, usersa, passwordyour_password, databaseTestDB) cursor conn.cursor(as_dictTrue) # 单条更新 cursor.execute(UPDATE Persons SET Age 21 WHERE Name 张三) # 批量更新 update_data [(李四, 26), (王五, 31)] cursor.executemany(UPDATE Persons SET Age %s WHERE Name %s, update_data) conn.commit() # 验证更新 cursor.execute(SELECT * FROM Persons) for row in cursor.fetchall(): print(f{row[Name]}的新年龄: {row[Age]}) conn.close()3.4 删除数据Delete删除操作同样支持单条和批量conn pymssql.connect(serverlocalhost, usersa, passwordyour_password, databaseTestDB) cursor conn.cursor(as_dictTrue) # 先查看当前数据 cursor.execute(SELECT * FROM Persons) print(删除前) for row in cursor.fetchall(): print(row) # 单条删除 cursor.execute(DELETE FROM Persons WHERE Name 张三) # 批量删除 delete_ids [(2), (3)] # 假设要删除ID为2和3的记录 cursor.executemany(DELETE FROM Persons WHERE ID %s, delete_ids) conn.commit() # 验证删除 cursor.execute(SELECT * FROM Persons) print(\n删除后) for row in cursor.fetchall(): print(row) conn.close()4. 中文编码问题解决方案在实际使用中中文乱码是个常见问题。这是因为SQL Server默认使用Latin1编码而我们需要处理中文。有几种解决方案修改数据库字段类型为NVARCHARALTER TABLE Persons ALTER COLUMN Name NVARCHAR(50)在Python代码中手动转码for row in cursor.fetchall(): name row[Name].encode(latin1).decode(gbk) print(f姓名: {name})连接时指定字符集conn pymssql.connect(serverlocalhost, usersa, passwordyour_password, databaseTestDB, charsetutf8)我推荐第一种方法一劳永逸地解决问题。如果无法修改数据库结构再考虑后两种方案。5. 封装可复用的数据库操作类为了在实际项目中更方便地使用我们可以把数据库操作封装成一个类import pymssql class MSSQLHelper: def __init__(self, server, user, password, database): self.server server self.user user self.password password self.database database self.conn None def __enter__(self): self.connect() return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() def connect(self): 建立数据库连接 self.conn pymssql.connect( serverself.server, userself.user, passwordself.password, databaseself.database, charsetutf8 ) def close(self): 关闭连接 if self.conn: self.conn.close() def execute_query(self, sql, paramsNone): 执行查询语句返回结果列表 with self.conn.cursor(as_dictTrue) as cursor: cursor.execute(sql, params) return cursor.fetchall() def execute_non_query(self, sql, paramsNone): 执行非查询语句返回影响行数 with self.conn.cursor() as cursor: cursor.execute(sql, params) self.conn.commit() return cursor.rowcount def execute_scalar(self, sql, paramsNone): 执行查询返回第一行第一列的值 with self.conn.cursor() as cursor: cursor.execute(sql, params) return cursor.fetchone()[0] def batch_insert(self, table, columns, data): 批量插入数据 placeholders , .join([%s] * len(columns)) columns_str , .join(columns) sql fINSERT INTO {table} ({columns_str}) VALUES ({placeholders}) with self.conn.cursor() as cursor: cursor.executemany(sql, data) self.conn.commit() return cursor.rowcount # 使用示例 with MSSQLHelper(localhost, sa, your_password, TestDB) as db: # 查询 results db.execute_query(SELECT * FROM Persons WHERE Age %s, (25,)) for row in results: print(row) # 插入 db.execute_non_query(INSERT INTO Persons (Name, Age) VALUES (%s, %s), (钱七, 40)) # 批量插入 data [(孙八, 45), (周九, 50)] db.batch_insert(Persons, [Name, Age], data)这个封装类具有以下特点支持上下文管理器with语句自动管理连接提供常用的查询、非查询、标量查询方法支持参数化查询防止SQL注入提供批量插入的便捷方法自动处理字符编码问题在实际项目中你可以根据需要扩展这个类比如添加事务支持、连接池等功能。