クイック スタート: Python 用 mssql-python ドライバーを使用した迅速なプロトタイプ作成

このクイックスタートでは、 Streamlit を使って迅速にレポートを作成し、ユーザーのフィードバックを早期に集めることができます。 Pythonのmssql-pythonドライバーを使ってデータベースに接続し、レポートに読み込まれたデータを読み取るだけです。

mssql-python ドライバーでは、Windows マシンへの外部依存関係は必要ありません。 ドライバーは、1 つの pip インストールで必要なすべてのものをインストールします。これにより、アップグレードとテストの時間がない他のスクリプトを中断することなく、新しいスクリプトに最新バージョンのドライバーを使用できます。

mssql-python のドキュメント | mssql-python ソース コード | パッケージ (PyPI) | Uv

[前提条件]

オペレーティングシステムをインストールするための前提条件

オペレーティング システム固有の 1 回限りの前提条件をインストールします。 Windowsユーザーはこのステップをスキップできます。 プラットフォームの詳細は 「Install 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 Databaseでは、接続文字列タブのODBCの接続文字列を使い、シナリオに合わせて認証タイプを調整してください。 接続文字列の構文の詳細については、「mssql-python 用の接続文字列」を参照してください。

    ヒント

    macOS では、 ActiveDirectoryInteractive と ActiveDirectoryDefault の両方が Microsoft Entra 認証で機能します。 ActiveDirectoryInteractive スクリプトを実行するたびにサインインするように求められます。 繰り返し表示されるサインイン プロンプトを避けるために、Azure CLIaz loginコマンドで一度サインインし、ActiveDirectoryDefaultを使いましょう。 ActiveDirectoryDefault キャッシュされた認証情報を再利用します。

uv run を使用してスクリプトを実行する

  1. 以前のターミナル ウィンドウで、または同じディレクトリに対して新しいターミナル ウィンドウを開き、次のコマンドを実行します。

    uv run streamlit run main.py
    

    レポートが Web ブラウザーの新しいタブで開きます。

  2. レポートを試して、そのしくみを確認してください。 何かを変更する場合は、 main.py 保存し、ブラウザー ウィンドウの右上隅にある再読み込みオプションを使用します。

  3. プロトタイプを共有するには、 .venv フォルダーを除くすべてのファイルを他のコンピューターにコピーします。 .venv フォルダーは、最初の実行で再作成されます。