快速入門:使用 Python 的 mssql-python 驅動程式快速建立原型

在這個快速入門中,你可以 Streamlit 快速建立報告,以便及早收集使用者回饋。 你用 mssql-python Python 的驅動程式連接資料庫,讀取已載入到報表中的資料。

在 Windows 電腦上,mssql-python 驅動程式不需要任何外部相依性。 驅動程式會透過單一 pip 安裝來安裝所需的所有內容,讓您可以將最新版本的驅動程式用於新腳本,而不會中斷您沒有時間升級和測試的其他腳本。

MSSQL-Python 文件 | MSSQL-Python 原始碼 | 套件(PyPI) | uv

先決條件

  • Python 3.10 或更新版本。

  • 如果你還沒有 Python,建議安裝 Python 執行環境 和 Pip 套件管理器 ,從 python.org 安裝。

  • 不想使用自己的系統環境? 遵循容器與本地開發,建立可重現的開發容器或 GitHub Codespaces 環境。

  • Visual Studio Code 使用下列擴充套件:

  • Azure Command-Line 介面(CLI)用於 macOS 和 Linux 的無密碼認證。

  • 如果你還沒有安裝<該軟體>,請依照 安裝說明進行操作。

  • SQL Server、Azure SQL Database 或 Fabric 中的 SQL 資料庫上的資料庫,具有 AdventureWorks2025 範例結構描述和有效的連接字串。

安裝作業系統先決條件

安裝一次性作業系統特定先決條件。 Windows 使用者可以跳過此步驟。 完整平台細節請參見 安裝 mssql-python。

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

建立 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 數值。

    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 Database 或 Fabric 中的 SQL 資料庫,請使用 連接字串s 標籤中的 ODBC 連接字串,並根據你的情境調整認證類型。 欲了解更多關於 連接字串 語法的資訊,請參閱 mssql-python 的 Connection strings。

    小提示

    在 macOS 上,兩者都ActiveDirectoryInteractiveActiveDirectoryDefault適用於 Microsoft Entra 認證。 ActiveDirectoryInteractive 每次執行腳本時都會提示你登入。 為避免重複出現登入提示,請使用 Azure CLIaz login 指令登入一次,然後使用 ActiveDirectoryDefault。 ActiveDirectoryDefault 重複使用快取的憑證。

使用 uv run 執行腳本

  1. 在先前的終端機視窗中,或開啟至相同目錄的新終端機視窗中,執行下列命令。

    uv run streamlit run main.py
    

    您的報告會在網頁瀏覽器的新索引標籤中開啟。

  2. 請嘗試您的報表,看看其運作方式。 如果您更改了任何內容,請保存 main.py 並使用瀏覽器窗口右上角的重新加載選項。

  3. 若要共用原型,請將資料夾以外的 .venv 所有檔案複製到另一台電腦。 .venv資料夾會在第一次執行時重新建立。