在本快速入門中,您可以使用來 Streamlit 快速建立報表,讓您快速收集使用者意見反應,以確保您走在正確的軌道上。您可以使用 mssql-python Python 驅動程式連線到資料庫,並讀取載入到報表中的資料。
在 Windows 電腦上,mssql-python 驅動程式不需要任何外部相依性。 驅動程式會透過單一 pip 安裝來安裝所需的所有內容,讓您可以將最新版本的驅動程式用於新腳本,而不會中斷您沒有時間升級和測試的其他腳本。
MSSQL-Python 文件 | MSSQL-Python 原始碼 | 套件(PyPI) | uv
先決條件
蟒蛇3
如果你還沒有 Python,建議安裝 Python 執行環境 和 Pip 套件管理器 ,從 python.org 安裝。
不想使用自己的系統環境? 使用 GitHub Codespaces 開啟為開發容器。
Visual Studio Code 使用下列擴充套件:
Azure Command-Line 介面(CLI)用於 macOS 和 Linux 的無密碼認證。
如果你還沒有安裝<該軟體>,請依照
安裝說明進行 操作。SQL Server、Azure SQL Database 或 Fabric 中的 SQL 資料庫上的資料庫,具有
AdventureWorks2025範例結構描述和有效的連接字串。安裝一次性作業系統特定先決條件。
建立 SQL 資料庫
此快速入門需使用 AdventureWorks2025 輕量級 架構,支援 Microsoft SQL Server、Fabric 中的 SQL 資料庫或 Azure SQL 資料庫。
建立專案並執行程式碼
建立新專案
在開發目錄中開啟命令提示字元。 如果您沒有,請建立一個名為 、
python等的新scripts目錄。避免在 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, 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)在匯入和
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找到這個代碼。
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 資料庫類型。 如果您要連線到 Azure SQL 資料庫 或 Fabric 中的 SQL 資料庫,請使用 [連接字串] 索引標籤中的 ODBC 連接字串。您可能需要根據您的案例調整驗證類型。 如需連接字串及其語法的詳細資訊,請參閱 連接字串語法參考。
使用 uv run 執行腳本
小提示
在 macOS 上,兩者都ActiveDirectoryInteractiveActiveDirectoryDefault適用於 Microsoft Entra 認證。
ActiveDirectoryInteractive 每次執行腳本時都會提示你登入。 為避免重複登入提示,請透過 Azure CLI 執行 az login,然後使用 ActiveDirectoryDefault,重複使用快取的憑證。
在先前的終端機視窗中,或開啟至相同目錄的新終端機視窗中,執行下列命令。
uv run streamlit run main.py您的報告會在網頁瀏覽器的新索引標籤中開啟。
請嘗試您的報表,看看其運作方式。 如果您更改了任何內容,請保存
main.py並使用瀏覽器窗口右上角的重新加載選項。若要共用原型,請將資料夾以外的
.venv所有檔案複製到另一台電腦。.venv資料夾會在第一次執行時重新建立。
後續步驟
請造訪 mssql-python 驅動程式 GitHub 存放庫以取得更多範例,以提供想法或報告問題。