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

2025-10-31

在本快速入门中，你将用于 Streamlit 快速创建报表，使你能够快速收集用户反馈，以确保你处于正确的轨道。使用用于 Python 的 mssql-python 驱动程序连接到数据库并读取加载到报表中的数据。

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

mssql-python 文档 | mssql-python 源代码 | 包（PyPi） | uv

先决条件

Python 3
- 如果还没有 Python，请从 python.org 安装 Python 运行时和 Python 包索引（PyPI）包管理器。
- 更倾向于不使用自己的环境吗？使用 GitHub Codespaces 以 devcontainer 身份打开。
具有以下扩展的 Visual Studio Code：
- 适用于 Visual Studio Code 的 Python 扩展
- （可选）Azure Repos
（可选）Azure Command-Line 接口（CLI）
如果尚未安装uv，请按照uv中的说明安装https://docs.astral.sh/uv/getting-started/installation/。
在 SQL Server、Azure SQL 数据库或 Fabric 中的 SQL 数据库上使用 AdventureWorks2022 示例架构和有效连接字符串的数据库。

安装一次性操作系统特定的先决条件。

apk add libtool krb5-libs krb5-dev

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

dnf install -y libtool-ltdl krb5-libs

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

zypper install -y libltdl7

brew install openssl

创建 SQL 数据库

本快速入门要求在 Microsoft SQL Server、Fabric 或 Azure SQL 数据库中的 SQL 数据库上使用 AdventureWorks2022 轻型 架构。

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

创建新项目

在开发目录中打开命令提示符。如果没有目录，请创建名为“pythonscripts等”的新目录。避免 OneDrive 上的文件夹，同步可能会干扰管理虚拟环境。

使用 . 创建一个新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

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

更新说明以更具描述性。

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

保存并关闭该文件。

更新 main.py

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

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

 if __name__ == "__main__":
   main()

在文件的顶部，在包含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
```

在导入和行 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] c inner join SalesLT.SalesOrderHeader soh on c.CustomerId = soh.CustomerId;"""

   df = load_data(SQL_QUERY)

   event = st.dataframe(
       df,
       width='stretch',
       hide_index=True,
       on_select="rerun",
       selection_mode="single-row"
   )

   customer = event.selection.rows

   if len(customer) == 0:
       SQL_QUERY = """select soh.OrderDate, SUM(sod.OrderQty), SUM(sod.OrderQty * sod.UnitPrice) as spend,  pc.Name as ProductCategory from SalesLT.SalesOrderDetail sod inner join SalesLt.SalesOrderHeader soh on sod.    salesorderid = soh.salesorderid inner join SalesLt.Product p on sod.productid = p.productid inner join SalesLT.ProductCategory pc on p.ProductCategoryID = pc.ProductCategoryID GROUP BY soh.OrderDate, pc.Name ORDER     BY soh.OrderDate, pc.Name;"""
   else:
       SQL_QUERY = f"""select soh.OrderDate, SUM(sod.OrderQty), SUM(sod.OrderQty * sod.UnitPrice) as spend,  pc.Name as ProductCategory from SalesLT.SalesOrderDetail sod inner join SalesLt.SalesOrderHeader soh on sod.    salesorderid = soh.salesorderid inner join SalesLt.Product p on sod.productid = p.productid inner join SalesLT.ProductCategory pc on p.ProductCategoryID = pc.ProductCategoryID where soh.CustomerID = {df.loc    [customer, 'CustomerID'].values[0]} GROUP BY soh.OrderDate, pc.Name ORDER BY soh.OrderDate, pc.Name;"""

   st.write("Here's a summary of spend by product category over time:")
   st.bar_chart(load_data(SQL_QUERY).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]}")
       SQL_QUERY = f"""SELECT * FROM [SalesLT].[SalesOrderHeader] soh  WHERE soh.CustomerID = {df.loc[customer, 'CustomerID'].values[0]};"""
       st.dataframe(load_data(SQL_QUERY), hide_index=True, width='stretch')
       SQL_QUERY = f"""SELECT sod.* FROM [SalesLT].[SalesOrderHeader] soh INNER JOIN SalesLT.SalesOrderDetail sod on soh.SalesOrderId = sod.SalesOrderId WHERE CustomerID = {df.loc[customer, 'CustomerID'].values[0]};"""
       st.dataframe(load_data(SQL_QUERY), hide_index=True, width='stretch')

在导入和 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) -> pd.DataFrame:
    data = pd.read_sql_query(SQL_QUERY, get_connection())
    return data

查找此代码。

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

将其替换为下面的代码。

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

保存并关闭 main.py。

保存连接字符串

打开.gitignore文件，并为.env文件添加一个排除项。文件应类似于此示例。完成后，请务必保存并关闭它。

# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env

在当前目录中，创建一个名为 .env 的新文件。
在 .env 文件中，为连接字符串 SQL_CONNECTION_STRING 添加一个条目。将此处的示例替换为实际连接字符串值。
```
SQL_CONNECTION_STRING="Server=<server_name>;Database={<database_name>};Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"
```
小窍门

此处使用的连接字符串在很大程度上取决于要连接到的 SQL 数据库的类型。如果要连接到 Fabric 中的 Azure SQL 数据库 或 SQL 数据库，请使用连接字符串选项卡中的 ODBC 连接字符串。可能需要根据方案调整身份验证类型。有关连接字符串及其语法的详细信息，请参阅连接字符串语法参考。

使用 uv run 命令执行脚本

小窍门

若要在 macOS 中使用 Microsoft Entra 身份验证，需要通过 Visual Studio Code 中的 Azure Repos 扩展或通过 az login运行来登录。

在之前所在的终端窗口中，或打开同一目录的新终端窗口，运行以下命令。
```
 uv run streamlit run main.py
```
报表将在 Web 浏览器中的新选项卡中打开。
尝试生成的报告，看看它是如何工作的。如果更改任何内容，请保存 main.py 并使用浏览器窗口右上角的重新加载选项。
若要共享原型，请将除文件夹以外的 .venv 所有文件复制到另一台计算机。第一次运行时将重新创建 .venv 文件夹。

后续步骤

访问mssql-python 驱动程序的 GitHub 存储库以获取更多示例，并贡献想法或报告问题。

GitHub 上的 mssql-python 驱动程序

反馈

此页面是否有帮助？