共用方式為


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

在本快速入門中,您可以使用來 Streamlit 快速建立報表,讓您快速收集使用者意見反應,以確保您走在正確的軌道上。您可以使用 mssql-python Python 驅動程式連線到資料庫,並讀取載入到報表中的資料。

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

mssql-python 文件 | mssql-python 原始程式碼 | 套件 (PyPi) | uv

先決條件


建立 SQL 資料庫

此快速入門需使用 AdventureWorks2025 輕量級 架構,支援 Microsoft SQL Server、Fabric 中的 SQL 資料庫或 Azure SQL 資料庫。

建立專案並執行程式碼

建立新專案

  1. 在開發目錄中開啟命令提示字元。 如果您沒有,請建立一個名為 、 python等的新scripts目錄。避免在 OneDrive 上放置資料夾,同步可能會幹擾虛擬環境的管理。

  2. 建立新的專案 使用uv

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

新增相依項目

在相同的目錄中,安裝 mssql-pythonstreamlitpython-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-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 資料庫類型。 如果您要連線到 Azure SQL 資料庫Fabric 中的 SQL 資料庫,請使用 [連接字串] 索引標籤中的 ODBC 連接字串。您可能需要根據您的案例調整驗證類型。 如需連接字串及其語法的詳細資訊,請參閱 連接字串語法參考

使用 uv run 執行腳本

小提示

若要在 macOS 中使用 Microsoft Entra 驗證,您必須透過 Visual Studio Code 中的 Azure Repos 延伸模組登入,或透過 az login 執行

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

     uv run streamlit run main.py
    
  2. 您的報告會在網頁瀏覽器的新索引標籤中開啟。

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

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

後續步驟

請造訪 mssql-python 驅動程式 GitHub 存放庫以取得更多範例,以提供想法或報告問題。