通过

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

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

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

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

先决条件

  • Python 3

  • 如果尚未安装uv，请按照uv中的说明安装https://docs.astral.sh/uv/getting-started/installation/。

  • 在 SQL Server、Azure SQL 数据库或 Fabric 中的 SQL 数据库上使用 AdventureWorks2022 示例架构和有效连接字符串的数据库。

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

    apk add libtool krb5-libs krb5-dev

创建 SQL 数据库

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

创建项目并运行代码

创建新项目

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

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

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

添加依赖项

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

uv add mssql-python python-dotenv streamlit

更新 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-protyping-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,
       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')

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

  5. 查找此代码。

    def main():
    print("Hello from rapid-protyping-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 命令执行脚本

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

     uv run streamlit run main.py

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

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

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

后续步骤

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

GitHub 上的 mssql-python 驱动程序