步骤 3：使用 pymssql 连接到 SQL 的概念证明

项目
08/28/2023

此示例概念证明使用 pymssql 连接到 SQL 数据库。此示例假设你使用的是 AdventureWorksLT 示例数据库。

注意

应只将此示例视为概念证明。为清楚起见，示例代码已简化，不一定代表 Microsoft 建议的最佳做法。

先决条件

Python 3
- 如果还没有 Python，请根据 python.org 安装 Python 运行时和 Python Package Index (PyPI) 包管理器。
- 不想使用自己的环境？使用 GitHub Codespaces 以 devcontainer 身份打开。
  - 。
来自 PyPI 的 pymssql 包。
SQL 数据库和凭据。

连接和查询数据

使用凭据连接到数据库。

创建名为 app.py 的新文件。

添加模块 docstring。

"""
Connects to a SQL database using pymssql
"""

导入 pymssql 包。
```
import pymssql
```

使用 pymssql.connect 功能连接 SQL 数据库。

conn = pymssql.connect(
    server='<server-address>',
    user='<username>',
    password='<password>',
    database='<database-name>',
    as_dict=True
)

执行查询

使用 SQL 查询字符串执行查询并分析结果。

为 SQL 查询字符串创建变量。

SQL_QUERY = """
SELECT 
TOP 5 c.CustomerID, 
c.CompanyName, 
COUNT(soh.SalesOrderID) AS OrderCount 
FROM 
SalesLT.Customer AS c 
LEFT OUTER JOIN SalesLT.SalesOrderHeader AS soh ON c.CustomerID = soh.CustomerID 
GROUP BY 
c.CustomerID, 
c.CompanyName 
ORDER BY 
OrderCount DESC;
"""

使用 cursor.execute 从数据库查询中检索结果集。
```
cursor = conn.cursor()
cursor.execute(SQL_QUERY)
```
注意

此函数实质上接受任意查询，并返回可使用 cursor.fetchone() 循环访问的结果集。

与 cursor.fetchall 循环一起使用 foreach，从数据库中获取所有记录。然后打印记录。

records = cursor.fetchall()
for r in records:
    print(f"{r['CustomerID']}\t{r['OrderCount']}\t{r['CompanyName']}")

保存 app.py 文件。

打开终端并测试应用程序。

python app.py

29485   1       Professional Sales and Service
29531   1       Remarkable Bike Store
29546   1       Bulk Discount Store
29568   1       Coalition Bike Company
29584   1       Futuristic Bikes

插入一行作为事务

此示例展示了安全地执行 INSERT 语句并传递参数。将参数作为值传递可保护应用程序不受 SQL 注入攻击。

从 randrange 库导入 random。
```
from random import randrange
```
生成随机的产品编号。
```
productNumber = randrange(1000)
```
提示

在此处生成随机的产品编号可确保可以多次运行此示例。

创建 SQL 语句字符串。

SQL_STATEMENT = """
INSERT SalesLT.Product (
Name, 
ProductNumber, 
StandardCost, 
ListPrice, 
SellStartDate
) OUTPUT INSERTED.ProductID 
VALUES (%s, %s, %s, %s, CURRENT_TIMESTAMP)
"""

使用 cursor.execute 执行该语句。

cursor.execute(
    SQL_STATEMENT,
    (
        f'Example Product {productNumber}', 
        f'EXAMPLE-{productNumber}', 
        100,
        200
    )
)

使用 cursor.fetchone 提取单个结果，打印结果的唯一标识符，然后使用 connection.commit 将该操作作为事务提交。
```
result = cursor.fetchone()
print(f"Inserted Product ID : {result['ProductID']}")
conn.commit()
```
提示

也可选择使用 connection.rollback 回滚该事务。
使用 cursor.close 和 connection.close 关闭游标和连接。
```
cursor.close()
conn.close()
```
保存 app.py 文件并再次测试应用程序
```
python app.py
```
```
Inserted Product ID : 1001
```

后续步骤

Python 开发人员中心。