mssql-python驱动为SQL查询执行、参数化查询、批处理和预备语句提供了光标方法。
基本查询执行
使用游标的 execute() 方法运行 SQL 语句：
import mssql_python
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE Color = 'Black'")
rows = cursor.fetchall()
for row in rows:
print(row.Name, row.ListPrice)
cursor.close()
conn.close()
参数化查询
始终使用参数化查询来防止 SQL 注入。 驱动程序的默认参数样式是 pyformat （命名占位符），但它也支持 qmark （位置占位符）。 对 ODBC {CALL} 转义序列使用 qmark。
Pyformat 风格（默认）
使用带 %(name)s 语法的命名占位符并传递词典：
cursor.execute(
"SELECT Name, ListPrice FROM Production.Product WHERE Color = %(color)s AND ListPrice > %(price)s",
{"color": "Black", "price": 10.00}
)
Qmark 风格
使用带有 ? 的位置占位符，并传递元组或列表：
cursor.execute(
"SELECT Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = ? AND ListPrice > ?",
(1, 10.00)
)
驱动程序会根据你的SQL查询和参数类型自动检测参数样式。
INSERT、UPDATE、DELETE操作
对于数据修改语句，应使用参数化查询并提交事务：
cursor.execute("CREATE TABLE #ExecDemo (Name NVARCHAR(50), CategoryID INT, Price DECIMAL(10,2))")
cursor.execute(
"INSERT INTO #ExecDemo (Name, CategoryID, Price) VALUES (%(name)s, %(category)s, %(price)s)",
{"name": "New Product", "category": 1, "price": 19.99}
)
conn.commit()
print(f"Rows affected: {cursor.rowcount}")
使用 executemany() 进行批量执行
使用 executemany() 高效地插入多行。 驱动程序采用列级参数绑定以实现高性能：
products = [
{"name": "Product A", "category": 1, "price": 10.00},
{"name": "Product B", "category": 1, "price": 15.00},
{"name": "Product C", "category": 2, "price": 20.00},
]
cursor.execute("CREATE TABLE #BatchDemo (Name NVARCHAR(50), CategoryID INT, Price DECIMAL(10,2))")
cursor.executemany(
"INSERT INTO #BatchDemo (Name, CategoryID, Price) VALUES (%(name)s, %(category)s, %(price)s)",
products
)
conn.commit()
print(f"Rows inserted: {cursor.rowcount}")
采用 QMARK 风格：
products = [
("Product A", 1, 10.00),
("Product B", 1, 15.00),
("Product C", 2, 20.00),
]
cursor.execute("CREATE TABLE #QmarkDemo (Name NVARCHAR(50), CategoryID INT, Price DECIMAL(10,2))")
cursor.executemany(
"INSERT INTO #QmarkDemo (Name, CategoryID, Price) VALUES (?, ?, ?)",
products
)
conn.commit()
多语句批量执行
在连接上使用 batch_execute() ，在一次调用中执行多个不同的语句：
results, cursor = conn.batch_execute(
[
"CREATE TABLE #BatchExec (Name NVARCHAR(50), CategoryID INT)",
"INSERT INTO #BatchExec (Name, CategoryID) VALUES (%(name)s, %(cat)s)",
"SELECT COUNT(*) FROM #BatchExec"
],
[
None, # No params for CREATE
{"name": "New Item", "cat": 1}, # Params for INSERT
None # No params for SELECT
]
)
print(f"CREATE result: {results[0]}")
print(f"INSERT affected: {results[1]} rows")
print(f"Row count: {results[2][0][0]}")
预定义语句
驱动程序默认准备查询（use_prepare=True）。 当你在同一光标上多次执行同一个SQL字符串时，驱动会自动在后续调用中重复使用已准备好的语句：
# First execution prepares the statement
cursor.execute(
"SELECT Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = %(subcategory_id)s",
{"subcategory_id": 1},
)
rows1 = cursor.fetchall()
# Same SQL string on same cursor → driver reuses the prepared plan
cursor.execute(
"SELECT Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = %(subcategory_id)s",
{"subcategory_id": 2},
)
rows2 = cursor.fetchall()
跳过准备，改用直接执行：
cursor.execute(
"SELECT Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = 1",
use_prepare=False # Uses SQLExecDirectW instead of SQLPrepareW
)
连接级执行
对于简单的一次性查询，直接对连接使用 execute() ：
# Creates cursor, executes, and returns cursor
cursor = conn.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
rows = cursor.fetchall()
cursor.close()
存储过程
通过使用 EXECUTE 或 ODBC {CALL} 转义语法调用存储过程。 有关输出参数、多结果集和事务模式的信息，请参见 存储过程。
cursor.execute(
"EXECUTE dbo.uspGetManagerEmployees @BusinessEntityID = %(business_entity_id)s",
{"business_entity_id": 16}
)
rows = cursor.fetchall()
设置输入大小
用于 setinputsizes() 显式声明参数类型，这可以提升批处理操作的性能：
cursor.setinputsizes([
(mssql_python.SQL_WVARCHAR, 50, 0), # NVARCHAR(50)
(mssql_python.SQL_INTEGER, 0, 0), # INT
])
cursor.executemany(
"SELECT ProductID, Name FROM Production.Product WHERE Name LIKE ? AND ProductSubcategoryID = ?",
[("Road%", 2), ("Mountain%", 1)]
)
注释
并非所有SQL类型常量都支持 setinputsizes()。
SQL_WVARCHAR 和 SQL_INTEGER 都很可靠。 对于十进制值，使用驱动程序的自动类型推断，而不是 SQL_DECIMAL，后者存在已知问题（GitHub #503）。
错误处理
将数据库操作放在 try-except 代码块中：
try:
cursor.execute("CREATE TABLE #ErrDemo (Name NVARCHAR(50) NOT NULL)")
cursor.execute("INSERT INTO #ErrDemo (Name) VALUES (%(name)s)", {"name": None})
conn.commit()
except mssql_python.IntegrityError as e:
print(f"Constraint violation: {e}")
conn.rollback()
except mssql_python.ProgrammingError as e:
print(f"SQL error: {e}")
conn.rollback()
最佳做法
- 始终使用参数化查询 以防止SQL注入。
-
对于批量插入，应使用 批量复制，而不是多次调用
execute()。 - 禁用自动提交时，显式提交事务。
- 完成后，关闭游标和连接以释放资源。
- 使用上下文管理器 进行自动资源清理：
with mssql_python.connect(connection_string) as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT TOP 5 Name, ListPrice FROM Production.Product")
rows = cursor.fetchall()
# Connection and cursor automatically closed