使用 Python 和 pyodbc 驱动程序连接到 Azure SQL 数据库并执行查询

适用于：Azure SQL 数据库

本快速入门教程介绍了如何使用 Python 和 Python SQL 驱动程序 - pyodbc 将应用程序连接到 Azure SQL 数据库中的数据库并执行查询。 本快速入门按照建议的无密码方法连接到数据库。 可以在无密码中心了解有关无密码连接的详细信息。

先决条件

• 一个 Azure 订阅。
• 配置了 Microsoft Entra 身份验证的 Azure SQL 数据库。 可以按照创建数据库快速入门创建数据库。
• 最新版本的 Azure CLI。
• 带 Python 扩展的 Visual Studio Code。
• Python 3.8 或更高版本。 如果使用 Linux 客户端计算机，请参阅安装 ODBC 驱动程序。

配置数据库

要与 Azure SQL 数据库建立安全的无密码连接，需要特定的数据库配置。 确认 Azure 中的逻辑服务器上的以下设置，在本地和托管环境中正确连接到 Azure SQL 数据库：

1. 对于本地开发连接，请确保逻辑服务器配置为允许本地计算机 IP 地址和其他 Azure 服务进行连接：

   • 导航到服务器的“网络”页。

   • 切换“所选网络”单选按钮以显示其他配置选项。

   • 选择“添加客户端 IPv4 地址(xx.xx.xx.xx)”以添加防火墙规则，这会启用来自本地计算机 IPv4 地址的连接。 或者，还可以选择“+ 添加防火墙规则”，输入所选的特定 IP 地址。

   • 确保选中“允许 Azure 服务和资源访问此服务器”复选框。

     

     警告

     对于生产方案，不建议启用“允许 Azure 服务和资源访问此服务器”设置。 实际应用程序应实现更安全的方法，例如更强的防火墙限制或虚拟网络配置。

     若要详细了解数据库安全配置，可以阅读以下资源：

     • 配置 Azure SQL 数据库防火墙规则。
     • 使用专用终结点配置虚拟网络。

2. 还必须为服务器启用 Microsoft Entra 身份验证，并为其分配 Microsoft Entra 管理员帐户。 对于本地开发连接，Microsoft Entra 管理员帐户应该是你也可以在本地用于登录到 Visual Studio 或 Azure CLI 的帐户。 可以在逻辑服务器的“Microsoft Entra ID”页上验证是否为服务器启用了 Microsoft Entra 身份验证。

   

3. 如果你使用个人 Azure 帐户，请确保已为 Azure SQL 数据库设置并配置 Microsoft Entra，以便将你的帐户分配为服务器管理员。如果你使用公司帐户，则很可能已为你配置了 Microsoft Entra ID。

创建项目

使用 Visual Studio Code 创建新的 Python 项目。

1. 打开 Visual Studio Code 并为项目创建一个新文件夹，并将目录更改为此文件夹。

   mkdir python-sql-azure
   cd python-sql-azure
   

2. 为应用创建一个虚拟环境。

   • Windows
   • macOS/Linux

   py -m venv .venv
   .venv\scripts\activate
   

3. 创建一个新的名为 app.py 的 Python 文件。

安装 pyodbc 驱动程序

若要使用 Python 连接到 Azure SQL 数据库，请安装 pyodbc 驱动程序。 此包充当数据提供程序，用于连接到数据库、执行命令和检索结果。 在本快速入门中，还将安装 flask、uvicorn 和 pydantic 包来创建和运行 API。

有关在所有操作系统上安装 pyodbc 驱动程序的详细信息和具体说明，请参阅为 pyodbc Python 开发配置开发环境。

1. 使用以下行创建 requirements.txt文件：

   pyodbc
   fastapi
   uvicorn[standard]
   pydantic
   azure-identity
   

2. 安装这些要求。

   pip install -r requirements.txt
   

配置本地连接字符串

对于本地开发和连接到 Azure SQL 数据库，请添加下面的 AZURE_SQL_CONNECTIONSTRING 环境变量。 将 <database-server-name> 和 <database-name> 占位符替换为你自己的值。 为 Bash shell 显示了示例环境变量。

在本地运行时，交互式身份验证提供无密码选项。 建议使用此选项，因为无需在本地系统上存储或管理身份验证机密。

export AZURE_SQL_CONNECTIONSTRING='Driver={ODBC Driver 18 for SQL Server};Server=tcp:<database-server-name>.database.chinacloudapi.cn,1433;Database=<database-name>;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30'

export AZURE_SQL_CONNECTIONSTRING='Driver={ODBC Driver 18 for SQL Server};Server=tcp:<database-server-name>.database.chinacloudapi.cn,1433;Database=<database-name>;UID=<user-name>;PWD=<user-password>;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30'

可以获取详细信息以从 Azure 门户创建连接字符串：

1. 转到“Azure SQL 服务器”，选择“SQL 数据库”页以查找数据库名称，然后选择数据库。

2. 在数据库上，转到“连接字符串”页以获取连接字符串信息。 在“ODBC”选项卡下查看。

添加代码以连接到 Azure SQL 数据库

在项目文件夹中，创建 app.py 文件并添加示例代码。 此代码创建一个 API，用于：

• 从环境变量中检索 Azure SQL 数据库连接字符串。
• 在启动期间，在数据库中创建 Persons 表（仅限测试场景）。
• 定义一个函数，用于从数据库中检索所有 Person 记录。
• 定义一个函数，用于从数据库中检索一个 Person 记录。
• 定义一个函数，用于从数据库中检索新的 Person 记录。

import os
import pyodbc, struct
from azure import identity

from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel

class Person(BaseModel):
    first_name: str
    last_name: Union[str, None] = None
    
connection_string = os.environ["AZURE_SQL_CONNECTIONSTRING"]

app = FastAPI()

@app.get("/")
def root():
    print("Root of Person API")
    try:
        conn = get_conn()
        cursor = conn.cursor()

        # Table should be created ahead of time in production app.
        cursor.execute("""
            CREATE TABLE Persons (
                ID int NOT NULL PRIMARY KEY IDENTITY,
                FirstName varchar(255),
                LastName varchar(255)
            );
        """)

        conn.commit()
    except Exception as e:
        # Table may already exist
        print(e)
    return "Person API"

@app.get("/all")
def get_persons():
    rows = []
    with get_conn() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM Persons")

        for row in cursor.fetchall():
            print(row.FirstName, row.LastName)
            rows.append(f"{row.ID}, {row.FirstName}, {row.LastName}")
    return rows

@app.get("/person/{person_id}")
def get_person(person_id: int):
    with get_conn() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM Persons WHERE ID = ?", person_id)

        row = cursor.fetchone()
        return f"{row.ID}, {row.FirstName}, {row.LastName}"

@app.post("/person")
def create_person(item: Person):
    with get_conn() as conn:
        cursor = conn.cursor()
        cursor.execute(f"INSERT INTO Persons (FirstName, LastName) VALUES (?, ?)", item.first_name, item.last_name)
        conn.commit()

    return item

def get_conn():
    credential = identity.DefaultAzureCredential(exclude_interactive_browser_credential=False)
    token_bytes = credential.get_token("https://database.chinacloudapi.cn/.default").token.encode("UTF-16-LE")
    token_struct = struct.pack(f'<I{len(token_bytes)}s', len(token_bytes), token_bytes)
    SQL_COPT_SS_ACCESS_TOKEN = 1256  # This connection option is defined by microsoft in msodbcsql.h
    conn = pyodbc.connect(connection_string, attrs_before={SQL_COPT_SS_ACCESS_TOKEN: token_struct})
    return conn

在本地运行并测试应用

应用已准备好在本地进行测试。

1. 在 Visual Studio Code 中运行 app.py 文件。

   uvicorn app:app --reload
   

2. 在应用 http://127.0.0.1:8000/docs 的 Swagger UI 页上，展开 POST 方法并选择“试用”。

   还可以使用试用 /redoc 查看另一种形式的 API 生成文档。

3. 修改示例 JSON 以包含名字和姓氏的值。 选择“执行”，将新记录添加到数据库中。 API 返回成功的响应。

4. 在 Swagger UI 页上，展开 GET 方法并选择“试用”。 选择“执行”，此时会返回你刚刚创建的人员。

部署到 Azure 应用服务

应用已准备好部署到 Azure。

1. 创建 start.sh 文件，以便 Azure 应用服务中的 gunicorn 可以运行 uvicorn。 start.sh 有一行：

   gunicorn -w 4 -k uvicorn.workers.UvicornWorker app:app
   

2. 使用 az webapp up 将代码部署到应用服务。 （可以使用选项 -dryrun 在不创建资源的情况下查看命令的作用。）

   az webapp up \
       --resource-group <resource-group-name> \
       --name <web-app-name>         
   

3. 使用 az webapp config set 命令将应用服务配置为使用 start.sh 文件。

   az webapp config set \
       --resource-group <resource-group-name> \
       --name <web-app-name> \
       --startup-file start.sh
   

4. 使用 az webapp identity assign 命令为应用服务启用系统分配的托管标识。

   az webapp identity assign \
       --resource-group <resource-group-name> \
       --name <web-app-name>
   

   在本快速入门中，使用系统分配的托管标识进行演示。 在更广泛的方案中，用户分配的托管标识更高效。 有关详细信息，请参阅托管标识最佳做法建议。 有关将用户分配的托管标识与 pyodbc 结合使用的示例，请参阅迁移 Python 应用程序以将无密码连接用于 Azure SQL 数据库。

将应用服务连接到 Azure SQL 数据库

在“配置数据库”部分中，为 Azure SQL 数据库服务器配置了网络和 Microsoft Entra 身份验证。 在本部分，你将完成数据库配置，并使用连接字符串配置应用服务以访问数据库服务器。

若要运行这些命令，可以使用任何可以连接到 Azure SQL 数据库的工具或 IDE，包括 SQL Server Management Studio (SSMS) 、Azure Data Studio 以及包含 SQL Server mssql 扩展的 Visual Studio Code。 此外，还可以按照快速入门：使用 Azure 门户查询编辑器查询 Azure SQL 数据库中所述使用 Azure 门户。

1. 使用 SQL 命令将用户添加到 Azure SQL 数据库，以创建用于无密码访问的用户和角色。

   CREATE USER [<web-app-name>] FROM EXTERNAL PROVIDER
   ALTER ROLE db_datareader ADD MEMBER [<web-app-name>]
   ALTER ROLE db_datawriter ADD MEMBER [<web-app-name>]
   

   有关详细信息，请参阅 包含的数据库用户 - 使你的数据库可移植。 有关显示相同原则但应用于 Azure VM 的示例，请参阅教程：使用 Windows VM 系统分配的托管标识访问 Azure SQL。 有关分配的角色的详细信息，请参阅固定数据库角色。

   如果禁用然后启用应用服务系统分配的托管标识，请删除用户并重新创建它。 运行 DROP USER [<web-app-name>] 并重新运行 CREATE 和 ALTER 命令。 若要查看用户，请使用 SELECT * FROM sys.database_principals。

2. 使用 az webapp config appsettings set 命令为连接字符串添加应用设置。

   az webapp config appsettings set \
       --resource-group <resource-group-name> \
       --name <web-app-name> \
       --settings AZURE_SQL_CONNECTIONSTRING="<connection-string>"
   

   对于部署的应用，连接字符串应类似于：

   Driver={ODBC Driver 18 for SQL Server};Server=tcp:<database-server-name>.database.chinacloudapi.cn,1433;Database=<database-name>;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30
   

   使用你自己的值填写 <dabaser-server-name> 和 <database-name>。

   无密码连接字符串不包含用户名或密码。 当应用在 Azure 中运行时，代码会使用 Azure 标识库中的 DefaultAzureCredential 来获取用于 pyodbc 的令牌。

测试已部署的应用程序

浏览到应用的 URL，测试与 Azure SQL 数据库的连接是否正常工作。 可以在应用服务的概述页上找到应用的 URL。

https://<web-app-name>.chinacloudsites.cn

将 /docs 追加到 URL 以查看 Swagger UI 并测试 API 方法。

恭喜！ 现在，你的应用程序已连接到本地和托管环境中的 Azure SQL 数据库。

相关内容

• 迁移 Python 应用程序以将无密码连接用于 Azure SQL 数据库 - 显示用户分配的托管标识。
• Azure 服务的无密码连接
• 托管标识的最佳做法建议