使用 Python 和 mssql-python 驱动程序连接到并查询Azure SQL Database

适用于：Azure SQL Database

本快速入门介绍如何在Azure SQL Database中将应用程序连接到数据库，并使用 Python 和 mssql-python 驱动程序执行查询。 mssql-python 驱动程序内置支持Microsoft Entra身份验证，使无密码连接变得简单。可以在无密码中心了解有关无密码连接的详细信息。

先决条件

Azure 订阅。
配有Microsoft Entra身份验证的 Azure SQL 数据库。可以使用 Quickstart：创建单一数据库 - Azure SQL Database。
使用配合Python 扩展的Visual Studio Code。
Python 3.10 或更高版本。
用于无密码身份验证的 Azure CLI（适用于 Windows、macOS 和 Linux）。

配置数据库

与Azure SQL Database的安全无密码连接需要某些数据库配置。在您的 Azure 逻辑服务器上验证以下设置，以在本地和托管环境中正常连接到 Azure SQL 数据库。

对于本地开发连接，请确保已将 Azure SQL 逻辑服务器配置为允许本地计算机 IP 地址和其他 Azure 服务进行连接：
1. 在 Azure 门户的资源菜单中，在 “安全性”下，选择“ 网络”。
2. 选择 “所选网络 ”按钮以显示其他配置选项。
3. 选择“添加客户端 IPv4 地址(xx.xx.xx.xx)”以添加防火墙规则，这会启用来自本地计算机 IPv4 地址的连接。或者，还可以选择“+ 添加防火墙规则”，输入所选的特定 IP 地址。
4. 确保选中 允许 Azure 服务和资源访问此服务器复选框。
  警告
  
  不建议在生产环境中启用 允许 Azure 服务和资源访问此服务器设置，因为这样做不符合安全最佳实践。实际应用程序应实现更安全的方法，例如更强的防火墙限制或虚拟网络配置。
  
  若要详细了解数据库安全配置，可以阅读以下资源：
  - 配置Azure SQL Database防火墙规则。
  - 使用专用终结点配置虚拟网络。
服务器还必须启用Microsoft Entra身份验证，并分配了Microsoft Entra管理员帐户。对于本地开发连接，Microsoft Entra 管理员帐户应该是一个您也可以在本地登录 Visual Studio 或 Azure CLI 使用的帐户。可以在逻辑服务器的 Microsoft Entra ID 页上验证服务器是否已启用Microsoft Entra身份验证。
如果使用个人 Azure 帐户，请确保已为 Azure SQL 数据库设置和配置 Microsoft Entra，以便将帐户分配为服务器管理员。如果使用公司帐户，Microsoft Entra ID 很可能已为你配置。

创建项目

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

打开Visual Studio Code，为项目创建新文件夹，并更改为该目录。
```
mkdir python-sql-azure
cd python-sql-azure
```

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

Windows
macOS/Linux

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

python3 -m venv .venv
source .venv/bin/activate

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

安装 mssql-python 驱动程序

若要使用Python连接到Azure SQL Database，请安装 mssql-python 驱动程序。此驱动程序内置了对Microsoft Entra身份验证的支持，无需手动处理令牌。在本快速入门中，还将安装 fastapi、uvicorn 和 pydantic 包来创建和运行 API。

注意

在 macOS 和 Linux 上，安装 mssql-python前需要系统依赖项。有关特定于平台的说明，请参阅安装 mssql-python 包。

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

mssql-python
fastapi
uvicorn[standard]
pydantic
python-dotenv

安装必要组件。
```
pip install -r requirements.txt
```

配置本地连接字符串

对于本地开发，请在项目文件夹中创建一个 .env 文件来存储connection string。这会使凭据远离代码和版本控制。

在项目文件夹中，创建一个名为 .env的文件。
将你的连接字符串添加到 AZURE_SQL_CONNECTIONSTRING 变量中。将 <database-server-name> 和 <database-name> 占位符替换为你自己的值。

mssql-python 驱动程序具有对Microsoft Entra身份验证的内置支持。使用 Authentication 参数指定身份验证方法。

ActiveDirectoryDefault 自动发现来自多个源的凭据，而无需交互式登录。这是 用于本地开发的推荐选项。

若要获得最可靠的本地开发体验，请先使用Azure CLI登录：

az login

然后在 .env 文件中使用此连接字符串格式：

AZURE_SQL_CONNECTIONSTRING=Server=<database-server-name>.database.chinacloudapi.cn;Database=<database-name>;Authentication=ActiveDirectoryDefault;Encrypt=yes;TrustServerCertificate=no;

ActiveDirectoryDefault 按以下顺序评估凭据：

环境变量 （对于服务主体凭据）
Managed identity（在Azure运行时）
Azure CLI （从 az login）
Visual Studio （仅限Windows）
Azure PowerShell （从 Connect-AzAccount -Environment AzureChinaCloud）

小窍门

对于生产应用程序，请使用方案的特定身份验证方法以避免凭据发现延迟：

Azure App Service/Functions：使用 ActiveDirectoryMSI （托管身份）
交互式用户登录：使用 ActiveDirectoryInteractive
服务主体：使用 ActiveDirectoryServicePrincipal

Microsoft Entra Interactive Authentication 使用多重身份验证技术设置连接。在此模式下，将显示“Azure身份验证”对话框，并允许输入凭据来完成连接。

AZURE_SQL_CONNECTIONSTRING=Server=<database-server-name>.database.chinacloudapi.cn;Database=<database-name>;Authentication=ActiveDirectoryInteractive;Encrypt=yes;TrustServerCertificate=no;

注意

在 macOS 上, ActiveDirectoryInteractive 和 ActiveDirectoryDefault 都可用于 Microsoft Entra 身份验证。 ActiveDirectoryInteractive 每次运行脚本时都会提示你登录。若要避免重复的登录提示，请通过 Azure CLI 运行 az login 进行一次登录，然后使用 ActiveDirectoryDefault 来重用缓存的凭据。

可以使用用户名和密码直接向SQL Server实例进行身份验证。

AZURE_SQL_CONNECTIONSTRING=Server=<database-server-name>.database.chinacloudapi.cn;Database=<database-name>;UID=<user-name>;PWD=<user-password>;Encrypt=yes;TrustServerCertificate=no;

警告

管理包含用户名、密码或访问密钥等机密的连接字符串时要小心。不应将这些机密提交到源代码管理中，也不应放置在可能由意外用户访问的不安全位置。将.env添加到.gitignore文件中，以防止意外提交机密。

可以从Azure门户获取创建connection string的详细信息：

转到 Azure SQL Server，选择 SQL 数据库页以查找数据库名称，然后选择数据库。
在数据库上，转到“ 概述 ”页以获取服务器名称。

添加代码以连接到Azure SQL Database

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

使用.env从python-dotenv文件加载配置。
从环境变量中检索 Azure SQL Database 的连接字符串。
在启动期间，在数据库中创建 Persons 表（仅限测试场景）。
定义一个函数，用于从数据库中检索所有 Person 记录。
定义一个函数，用于从数据库中检索一个 Person 记录。
定义一个函数，用于从数据库中检索新的 Person 记录。

from os import getenv
from typing import Union
from dotenv import load_dotenv
from fastapi import FastAPI
from pydantic import BaseModel
from mssql_python import connect

load_dotenv()

class Person(BaseModel):
    first_name: str
    last_name: Union[str, None] = None

connection_string = getenv("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("""
            IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'Persons')
            CREATE TABLE Persons (
                ID int NOT NULL PRIMARY KEY IDENTITY,
                FirstName varchar(255),
                LastName varchar(255)
            );
        """)

        conn.commit()
        conn.close()
    except Exception as e:
        # Table might 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("INSERT INTO Persons (FirstName, LastName) VALUES (?, ?)",
                       (item.first_name, item.last_name))
        conn.commit()

    return item

def get_conn():
    """Connect using mssql-python with built-in Microsoft Entra authentication."""
    conn = connect(connection_string)
    conn.setautocommit(True)
    return conn

警告

此示例代码显示了不应在生产代码中使用的原始 SQL 语句。请改用对象关系映射程序 (ORM) 包（例如 SqlAlchemy），该包可生成更安全的对象层来访问数据库。

在本地运行并测试应用

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

在 Visual Studio Code 中运行 app.py 文件。
```
uvicorn app:app --reload
```
在应用 http://127.0.0.1:8000/docs 的 Swagger UI 页上，展开 POST 方法并选择“试用”。

还可以尝试 /redoc 查看 API 的另一种生成文档形式。
修改示例 JSON 以包含名字和姓氏的值。选择“执行”，将新记录添加到数据库中。 API 返回成功的响应。
在 Swagger UI 页上展开GET方法，然后选择尝试。选择执行，然后刚刚创建的人员会被返回。

部署到 Azure 应用服务

应用已准备好部署到Azure。

创建 start.sh 文件，以便Azure App Service中的 gunicorn 可以运行 uvicorn。 start.sh 有一行：
```
gunicorn -w 4 -k uvicorn.workers.UvicornWorker app:app
```
使用 az webapp up 将代码部署到应用服务。（可以使用选项 -dryrun 在不创建资源的情况下查看命令的作用。）
```
az webapp up \
    --resource-group <resource-group-name> \
    --name <web-app-name>         
```

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

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

使用 az webapp identity assign 命令为应用服务启用系统分配的托管标识。
```
az webapp identity assign \
    --resource-group <resource-group-name> \
    --name <web-app-name>
```
在本快速入门中，使用一个系统分配的托管标识来进行演示。在更广泛的方案中，用户分配的托管标识更高效。有关详细信息，请参阅托管标识最佳做法建议。有关将用户分配的托管标识与 mssql-python 配合使用的示例，请参阅将 Python 应用程序迁移到使用 Azure SQL Database 无密码连接。

将应用服务连接到Azure SQL Database

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

若要运行这些命令，可以使用任何可连接到Azure SQL Database的工具或 IDE，包括 SQL Server Management Studio （SSMS），并使用 MSSQL 扩展Visual Studio Code。也可以使用 Azure 门户，如 Quickstart 中所述：使用 Azure 门户查询编辑器查询Azure SQL Database。

使用 SQL 命令将用户添加到Azure SQL Database，以创建用于无密码访问的用户和角色。
```
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 相同的原则的示例，请参阅 Tutorial：使用 Windows VM 系统分配的托管标识访问 Azure SQL。有关分配的角色的详细信息，请参阅固定数据库角色。

如果禁用并重新启用应用服务系统分配的托管标识，请先删除该用户，然后重新创建用户。运行 DROP USER [<web-app-name>] 并重新运行 CREATE 和 ALTER 命令。若要查看用户，请使用 SELECT * FROM sys.database_principals。
使用 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>"
```
对于已部署的应用，connection string应类似于：
```
Server=<database-server-name>.database.chinacloudapi.cn;Database=<database-name>;Authentication=ActiveDirectoryMSI;Encrypt=yes;TrustServerCertificate=no;
```
使用你自己的值填写 <database-server-name> 和 <database-name>。

不含用户名或密码的无密码连接字符串。相反，当应用在 Azure 中运行时，mssql-python 驱动程序使用 ActiveDirectoryMSI 身份验证模式自动使用应用服务的托管标识进行身份验证。

测试已部署的应用程序

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

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

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

恭喜！应用程序现已连接到本地和托管环境中的Azure SQL Database。

Last updated on 2026-04-03