快速入门：使用适用于 Python 的 mssql-python 驱动程序快速原型制作

在这个快速入门中，你可以 Streamlit 快速创建报告，以便提前收集用户反馈。 你使用适用于 Python 的 mssql-python 驱动程序来连接数据库，并读取已加载到报告中的数据。

驱动程序 mssql-python 不需要 Windows 计算机上的任何外部依赖项。 驱动程序通过单个 pip 安装来安装它所需的一切，使你能够将最新版本的驱动程序用于新脚本，同时不会影响那些没有时间升级和测试的其他脚本。

mssql-python 文档 | mssql-python 源代码 | 包 （PyPI） | Uv

先决条件

• Python 3.10 或更高版本。

• 如果还没有 Python，请从 python.org 安装 Python 运行时和 pip 包管理器。

• 不想使用您自己的环境吗？ 遵循容器和本地开发，创建一个可复现的开发容器或 GitHub Codespaces 环境。

• 具有以下扩展的 Visual Studio Code：

  • 适用于 Visual Studio Code 的 Python 扩展

• Azure Command-Line 接口（CLI） 用于在 macOS 和 Linux 上进行无密码身份验证。

• 如果尚未安装 uv，请按照 安装说明操作。

• 一个基于 SQL Server、Azure SQL 数据库或 Fabric 中的 SQL 数据库的数据库，包含 AdventureWorks2025 示例架构和有效的连接字符串。

安装操作系统所需的先决条件

安装一次性操作系统特定的先决程序。 Windows 用户可以跳过这一步。 完整平台详情请参见 “安装 mssql-python”。

高山

apk add libtool krb5-libs krb5-dev libstdc++

Debian/Ubuntu

apt-get install -y libltdl7 libkrb5-3 libgssapi-krb5-2

RHEL

dnf install -y libtool-ltdl krb5-libs

SUSE

zypper install -y libltdl7 libkrb5-3 libgssapi-krb5-2

openSUSE

zypper install -y libltdl7

macOS

brew install openssl

创建 SQL 数据库

在以下平台之一创建或连接SQL数据库：

Azure SQL 数据库

使用 Azure 门户在几分钟内创建 SQL 数据库

Fabric 中的 SQL 数据库

在 Microsoft Fabric 中加载 SQL 数据库中的 AdventureWorks 示例数据

Microsoft SQL Server

从安装向导安装 SQL Server

AdventureWorks 示例数据库

SQL Server 容器

在 Docker 中创建 SQL Server 容器

用 sqlcmd 创建 SQL Server 容器

使用适用于 VS Code 的 MSSQL 扩展创建 SQL Server 容器

创建项目并运行代码

• 创建新项目
• 添加依赖项
• 启动 Visual Studio Code
• 更新 pyproject.toml
• 更新 main.py
• 保存连接字符串
• 使用 uv 运行执行脚本

创建新项目

1. 在开发目录中打开命令提示符。 如果没有，可以创建一个新的目录，比如 python 或 scripts。 避免在OneDrive上放置文件夹，因为同步会影响管理虚拟环境。

2. 使用 . 创建一个新uv。

   uv init rapid-prototyping-qs
   cd rapid-prototyping-qs
   

添加依赖项

在同一目录中，安装mssql-python、streamlit和python-dotenv包。

uv add mssql-python python-dotenv streamlit

启动 Visual Studio Code

在同一目录中运行以下命令。

code .

更新 pyproject.toml

1. 在编辑器中打开 pyproject.toml 文件。 该文件包含您项目的元数据。

2. 更新 description 数值。

   description = "A quick example of rapid prototyping using the mssql-python driver and Streamlit."
   

3. 保存并关闭该文件。

更新 main.py

1. 打开名为main.py的文件。 它应类似于此示例。

   def main():
       print("Hello from rapid-prototyping-qs!")
   
   if __name__ == "__main__":
       main()
   

2. 在文件顶部，在带有 def main()的行前添加以下导入。

   小窍门

   如果 Visual Studio Code 在解决包时遇到问题，则需要 更新解释器以使用虚拟环境。

   from os import getenv
   from dotenv import load_dotenv
   from mssql_python import connect, Connection
   import pandas as pd
   import streamlit as st
   

3. 在导入和行 def main()之间添加以下代码。

   def page_load() -> None:
       st.set_page_config(
           page_title="View Data",
           page_icon=":bar_chart:",
           layout="wide",
           initial_sidebar_state="expanded"
       )
   
       st.title("AdventureWorksLT Customer Order History")
   
       SQL_QUERY = """
       SELECT c.*
       FROM SalesLT.Customer AS c
       INNER JOIN SalesLT.SalesOrderHeader AS soh
           ON c.CustomerID = soh.CustomerID;
       """
   
       df = load_data(SQL_QUERY)
   
       event = st.dataframe(
           df,
           use_container_width=True,
           hide_index=True,
           on_select="rerun",
           selection_mode="single-row"
       )
   
       customer = event.selection.rows
   
       SPEND_QUERY_ALL = """
       SELECT
           soh.OrderDate,
           SUM(sod.OrderQty),
           SUM(sod.OrderQty * sod.UnitPrice) AS spend,
           pc.Name AS ProductCategory
       FROM SalesLT.SalesOrderDetail AS sod
       INNER JOIN SalesLT.SalesOrderHeader AS soh
           ON sod.SalesOrderID = soh.SalesOrderID
       INNER JOIN SalesLT.Product AS p
           ON sod.ProductID = p.ProductID
       INNER JOIN SalesLT.ProductCategory AS pc
           ON p.ProductCategoryID = pc.ProductCategoryID
       GROUP BY soh.OrderDate, pc.Name
       ORDER BY soh.OrderDate, pc.Name;
       """
       SPEND_QUERY_CUSTOMER = """
       SELECT
           soh.OrderDate,
           SUM(sod.OrderQty),
           SUM(sod.OrderQty * sod.UnitPrice) AS spend,
           pc.Name AS ProductCategory
       FROM SalesLT.SalesOrderDetail AS sod
       INNER JOIN SalesLT.SalesOrderHeader AS soh
           ON sod.SalesOrderID = soh.SalesOrderID
       INNER JOIN SalesLT.Product AS p
           ON sod.ProductID = p.ProductID
       INNER JOIN SalesLT.ProductCategory AS pc
           ON p.ProductCategoryID = pc.ProductCategoryID
       WHERE soh.CustomerID = ?
       GROUP BY soh.OrderDate, pc.Name
       ORDER BY soh.OrderDate, pc.Name;
       """
       if len(customer) == 0:
           spend_df = load_data(SPEND_QUERY_ALL)
       else:
           customer_id = int(df.loc[customer, 'CustomerID'].values[0])
           spend_df = load_data(SPEND_QUERY_CUSTOMER, params=(customer_id,))
   
       st.write("Here's a summary of spend by product category over time:")
       st.bar_chart(spend_df.set_index('ProductCategory')['spend'], use_container_width=True)
   
       if len(customer) > 0:
           st.write(f"Displaying orders for Customer ID: {df.loc[customer, 'CustomerID'].values[0]}")
           customer_id = int(df.loc[customer, 'CustomerID'].values[0])
           SQL_QUERY = """
           SELECT *
           FROM SalesLT.SalesOrderHeader AS soh
           WHERE soh.CustomerID = ?;
           """
           st.dataframe(load_data(SQL_QUERY, params=(customer_id,)), hide_index=True, use_container_width=True)
           SQL_QUERY = """
           SELECT sod.*
           FROM SalesLT.SalesOrderHeader AS soh
           INNER JOIN SalesLT.SalesOrderDetail AS sod
               ON soh.SalesOrderID = sod.SalesOrderID
           WHERE soh.CustomerID = ?;
           """
           st.dataframe(load_data(SQL_QUERY, params=(customer_id,)), hide_index=True, use_container_width=True)
   

4. 在导入和 def page_load() -> None:之间，添加此代码。

   _connection = None
   
   def get_connection() -> Connection:
       global _connection
       if not _connection:
           load_dotenv()
           _connection = connect(getenv("SQL_CONNECTION_STRING"))
       return _connection
   
   @st.cache_data
   def load_data(SQL_QUERY, params=None) -> pd.DataFrame:
       data = pd.read_sql_query(SQL_QUERY, get_connection(), params=params)
       return data
   

5. 查找此代码。

   def main():
       print("Hello from rapid-prototyping-qs!")
   

6. 将其替换为下面的代码。

   def main() -> None:
       page_load()
       if _connection:
           _connection.close()
   

7. 保存并关闭 main.py。

保存连接字符串

1. 打开.gitignore文件，并为.env文件添加一个排除项。 你的文件应该类似于以下示例。

   # Python-generated files
   __pycache__/
   *.py[oc]
   build/
   dist/
   wheels/
   *.egg-info
   
   # Virtual environments
   .venv
   
   # Connection strings and secrets
   .env
   

2. 保存并关闭 .gitignore。

3. 在当前目录中，创建一个名为 .env 的新文件。

4. 在 .env 文件中，为连接字符串 SQL_CONNECTION_STRING 添加一个条目。 请使用实际值替换占位符。

   SQL_CONNECTION_STRING="Server=<server>;Database=<database>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"
   

   连接字符串 取决于你连接的 SQL 数据库类型。 对于Azure SQL 数据库或Fabric中的SQL数据库，使用连接字符串标签中的ODBC 连接字符串，并根据你的场景调整认证类型。 有关连接字符串语法的更多信息，请参见 适用于 mssql-python 的连接字符串。

   小窍门

   在 macOS 上, ActiveDirectoryInteractive 和 ActiveDirectoryDefault 都可用于 Microsoft Entra 身份验证。 ActiveDirectoryInteractive 每次运行脚本时都会提示你登录。 为避免重复登录提示，请使用 Azure CLIaz login 命令登录一次，并使用 ActiveDirectoryDefault。 ActiveDirectoryDefault 重复使用缓存的凭据。

使用 uv run 命令执行脚本

1. 在之前所在的终端窗口中，或打开同一目录的新终端窗口，运行以下命令。

   uv run streamlit run main.py
   

   报表将在 Web 浏览器中的新选项卡中打开。

2. 尝试运行报告，看看它是如何工作的。 如果更改任何内容，请保存 main.py 并使用浏览器窗口右上角的重新加载选项。

3. 若要共享原型，请将除文件夹以外的 .venv 所有文件复制到另一台计算机。 第一次运行时将重新创建 .venv 文件夹。

相关内容

• 快速入门：使用适用于 Python 的 mssql-python 驱动程序进行可重复部署
• 程序化构建连接字符串
• 用 mssql-python 执行查询