Rychlý start: Rychlé vytváření prototypů pomocí ovladače mssql-python pro Python

V tomto rychlém startu můžete Streamlit rychle vytvořit report, abyste mohli získat zpětnou vazbu od uživatelů včas. Používáte mssql-python ovladač pro Python k připojení k databázi a čtení dat načtených do reportu.

Ovladač mssql-python nevyžaduje žádné externí závislosti na počítačích s Windows. Ovladač nainstaluje všechno, co potřebuje, s jednou pip instalací, což vám umožní používat nejnovější verzi ovladače pro nové skripty bez přerušení jiných skriptů, které nemáte čas upgradovat a testovat.

Dokumentace mssql-python, zdrojový kód mssql-python, Balíček (PyPI) uv

Požadavky

  • Python 3.10 nebo novější.

  • Pokud ještě nemáte Python, nainstalujte Python runtime a správce balíčků pip z python.org.

  • Nechcete používat vlastní prostředí? Postupujte podle Kontejner a místní vývoj a vytvořte reprodukovatelné vývojové prostředí devcontainer nebo GitHub Codespaces.

  • Visual Studio Code s následujícími rozšířeními:

    • Rozšíření Pythonu pro Visual Studio Code
  • Azure Command-Line Interface (CLI) pro ověřování bez hesla v systému macOS a Linux.

  • Pokud ještě uvnemáte, postupujte podle pokynů k instalaci.

  • Databáze na SQL Serveru, Azure SQL Database nebo SQL databáze v Fabric s ukázkovým AdventureWorks2025 schématem a platným připojovacím řetězcem.

Nainstalujte požadavky operačního systému

Nainstalujte požadavky specifické pro jednorázový operační systém. Uživatelé Windows mohou tento krok přeskočit. Pro úplné podrobnosti o platformě viz Instalace mssql-python.

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

Vytvoření databáze SQL

Vytvořte nebo se připojte k SQL databázi na jedné z následujících platforem:

Vytvoření projektu a spuštění kódu

Vytvoření nového projektu

  1. Otevřete příkazový řádek ve vývojovém adresáři. Pokud žádný nemáte, vytvořte nový adresář, například python nebo scripts. Vyhněte se složkám na OneDrive, protože synchronizace může narušit správu virtuálního prostředí.

  2. Vytvořte nový projekt pomocí uvpříkazu .

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

Přidejte závislosti

Ve stejném adresáři nainstalujte balíčky mssql-python, streamlit a python-dotenv.

uv add mssql-python python-dotenv streamlit

Spusťte Visual Studio Code

Ve stejném adresáři spusťte následující příkaz.

code .

Aktualizace pyproject.toml

  1. Otevřete soubor v editoru pyproject.toml . Tento soubor obsahuje metadata vašeho projektu.

  2. Aktualizujte hodnotu description .

    description = "A quick example of rapid prototyping using the mssql-python driver and Streamlit."
    
  3. Uložte a zavřete soubor.

Aktualizace main.py

  1. Otevřete soubor s názvem main.py. Měl by se podobat tomuto příkladu.

    def main():
        print("Hello from rapid-prototyping-qs!")
    
    if __name__ == "__main__":
        main()
    
  2. Na vrchol souboru přidejte následující importy před řádek s def main().

    Návod

    Pokud má Visual Studio Code potíže s řešením balíčků, musíte interpreta aktualizovat tak, aby používal virtuální prostředí.

    from os import getenv
    from dotenv import load_dotenv
    from mssql_python import connect, Connection
    import pandas as pd
    import streamlit as st
    
  3. Mezi importy a řádek s def main(), přidejte následující kód.

    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. Mezi importy a def page_load() -> None:přidejte tento kód.

    _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. Vyhledejte tento kód.

    def main():
        print("Hello from rapid-prototyping-qs!")
    
  6. Nahraďte ho tímto kódem.

    def main() -> None:
        page_load()
        if _connection:
            _connection.close()
    
  7. Uložte a zavřete main.py.

Uložte řetězec připojení

  1. .gitignore Otevřete soubor a přidejte vyloučení souborů.env. Váš soubor by měl být podobný následujícímu příkladu.

    # Python-generated files
    __pycache__/
    *.py[oc]
    build/
    dist/
    wheels/
    *.egg-info
    
    # Virtual environments
    .venv
    
    # Connection strings and secrets
    .env
    
  2. Uložte a zavřete .gitignore.

  3. V aktuálním adresáři vytvořte nový soubor s názvem .env.

  4. .env V souboru přidejte položku pro připojovací řetězec s názvem SQL_CONNECTION_STRING. Zástupné symboly nahraďte skutečnými hodnotami.

    SQL_CONNECTION_STRING="Server=<server>;Database=<database>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"
    

    připojovací řetězec závisí na typu SQL databáze, ke které se připojujete. Pro Azure SQL Database nebo SQL databázi ve Fabric použijte ODBC připojovací řetězec z karty připojovací řetězecs a upravte typ autentizace pro váš scénář. Pro více informací o syntaxi připojovací řetězec viz Connection strings for mssql-python.

    Návod

    V systému macOS funguje ověřování pomocí Microsoft Entra jak s ActiveDirectoryInteractive, tak s ActiveDirectoryDefault. ActiveDirectoryInteractive vás vyzve k přihlášení při každém spuštění skriptu. Abychom se vyhnuli opakovaným výzvám k přihlášení, přihlaste se jednou příkazem Azure CLIaz login a použijte ActiveDirectoryDefault. ActiveDirectoryDefault znovu použije uložené přihlašovací údaje.

Použijte 'uv run' ke spuštění skriptu

  1. V okně terminálu před nebo v novém okně terminálu, které se otevře ve stejném adresáři, spusťte následující příkaz.

    uv run streamlit run main.py
    

    Vaše zpráva se otevírá na nové kartě ve webovém prohlížeči.

  2. Vyzkoušejte sestavu a podívejte se, jak funguje. Pokud něco změníte, uložte main.py a použijte možnost opětovného načtení v pravém horním rohu okna prohlížeče.

  3. Pokud chcete sdílet prototyp, zkopírujte všechny soubory s výjimkou .venv složky na druhý počítač. Složka .venv se znovu vytvoří při prvním spuštění.