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

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

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

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

先决条件

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

apk add libtool krb5-libs krb5-dev

创建 SQL 数据库

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

创建项目并运行代码

创建新项目

  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 = "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] c inner join SalesLT.SalesOrderHeader 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 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;"""
   SPEND_QUERY_CUSTOMER = """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 = ? 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] 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] soh INNER JOIN SalesLT.SalesOrderDetail sod on soh.SalesOrderId = sod.SalesOrderId WHERE 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. 在当前目录中，创建一个名为 .env 的新文件。

  3. 在 .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 上, ActiveDirectoryInteractive 和 ActiveDirectoryDefault 都可用于 Microsoft Entra 身份验证。 ActiveDirectoryInteractive 每次运行脚本时都会提示你登录。 为避免重复登录提示，请通过Azure CLI运行 az login，然后使用ActiveDirectoryDefault，该登录可重复使用缓存的凭证。

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

     uv run streamlit run main.py

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

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

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

后续步骤

通过这些文章继续深入学习：

快速启动：可重复部署

  • Last updated on