Share via


Python と pyodbc ドライバーを使用して Azure SQL Database に接続してクエリを実行する

適用対象:Azure SQL Database

このクイックスタートでは、Python と Python SQL ドライバー - pyodbc を使って Azure SQL Database 内のデータベースにアプリケーションを接続し、クエリを実行する方法について説明します。 このクイックスタートでは、データベースに接続するための推奨されるパスワードレス アプローチに従います。 パスワードレス接続について詳しくは、パスワードレス ハブに関する記事を参照してください。

前提条件

データベースを構成する

Azure SQL Database への安全なパスワードレス接続には、特定のデータベース構成が必要です。 Azure の論理サーバーで次の設定を確認し、ローカル環境とホスト環境の両方で Azure SQL Database に適切に接続します。

  1. ローカル開発の接続の場合は、ローカル コンピューターの IP アドレスやその他の Azure サービスで接続できるように論理サーバーが構成されていることを確認します。

    • お使いのサーバーの [ネットワーク] ページに移動します。

    • [選択されたネットワーク] ラジオ ボタンを切り替えて、追加の構成オプションを表示します。

    • [Add your client IPv4 address(xx.xx.xx.xx)] (クライアント IPv4 アドレスの追加 (xx.xx.xx.xx)) を選び、ローカル コンピューターの IPv4 アドレスからの接続を有効にするファイアウォール規則を追加します。 または、[+ Add a firewall rule] (ファイアウォール規則の追加) を選び、選んだ特定の IP アドレスを入力することもできます。

    • [Azure サービスおよびリソースにこのサーバーへのアクセスを許可する] チェックボックスがオンになっていることを確認します。

      ファイアウォール規則を構成する方法を示すスクリーンショット。

      警告

      [Azure サービスおよびリソースにこのサーバーへのアクセスを許可する] 設定を有効にすることは、運用環境のシナリオでは推奨されるセキュリティ プラクティスではありません。 実際のアプリケーションでは、より強力なファイアウォール制限や仮想ネットワーク構成など、より安全なアプローチを実装する必要があります。

      データベース セキュリティの構成について詳しくは、次のリソースを参照してください。

  2. また、サーバーでは Microsoft Entra 認証が有効になっており、Microsoft Entra 管理者アカウントが割り当てられている必要があります。 ローカル開発接続の場合、Microsoft Entra 管理者アカウントは、ローカルで Visual Studio または Azure CLI にログインできるアカウントである必要があります。 論理サーバーの Microsoft Entra ID ページで、サーバーで Microsoft Entra 認証が有効になっているかどうかを確認できます。

    Microsoft Entra 認証を有効にする方法を示すスクリーンショット。

  3. 個人の Azure アカウントを使用している場合は、アカウントをサーバー管理者として割り当てるために、Microsoft Entra がセットアップされ、Azure SQL Database 用に構成されていることを確認してください。企業アカウントを使用している場合は、Microsoft Entra ID がおそらくすでに構成済みになっています。

プロジェクトを作成する

Visual Studio Code を使用して新しい Python プロジェクトを作成します。

  1. Visual Studio Code を開き、プロジェクト用の新しいフォルダーを作成して、ディレクトリをそこに変更します。

    mkdir python-sql-azure
    cd python-sql-azure
    
  2. アプリ用の仮想環境を作成します。

    py -m venv .venv
    .venv\scripts\activate
    
  3. app.py という名前の新しい Python ファイルを作成します。

pyodbc ドライバーをインストールする

Python を使って Azure SQL Database に接続するには、pyodbc ドライバーをインストールします。 このパッケージは、データベースへの接続、コマンドの実行、結果の取得を行うためのデータ プロバイダーとして機能します。 このクイックスタートでは、API を作成して実行するために、flaskuvicornpydantic パッケージもインストールします。

すべてのオペレーティング システムに pyodbc ドライバーをインストールするための詳しく具体的な手順については、「pyodbc Python 開発用に開発環境を構成する」をご覧ください。

  1. 次の行を含む requirements.txt ファイルを作成します。

    pyodbc
    fastapi
    uvicorn[standard]
    pydantic
    azure-identity
    
  2. 要件をインストールします。

    pip install -r requirements.txt
    

ローカル接続文字列を構成する

ローカル開発と Azure SQL Database への接続のため、次の AZURE_SQL_CONNECTIONSTRING 環境変数を追加します。 プレースホルダー <database-server-name><database-name> を実際の値に置き換えます。 示されている環境変数の例は、Bash シェルのものです。

ローカル環境で実行している場合、対話型認証ではパスワードレス オプションが提供されます。

Windows では、Microsoft Entra Interactive Authentication で Microsoft Entra 多要素認証テクノロジを使用して接続を設定できます。 このモードでは、サインイン ID を指定することで Azure 認証ダイアログがトリガーされ、ユーザーはパスワードを入力して接続を完了できます。

export AZURE_SQL_CONNECTIONSTRING='Driver={ODBC Driver 18 for SQL Server};Server=tcp:<database-server-name>.database.windows.net,1433;Database=<database-name>;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30'

詳細情報については、「ODBC ドライバーでの Microsoft Entra ID の使用」をご覧ください。 このオプションを使う場合は、資格情報の入力を求めるウィンドウを探します。

接続文字列を作成するための詳細は、Azure portal から入手できます。

  1. Azure SQL Server に移動し、[SQL データベース] ページを選んでデータベース名を検索して、データベースを選びます。

  2. データベースで [接続文字列] ページに移動して、接続文字列の情報を取得します。 [ODBC] タブの下を見てください。

Note

Azure Arc をインストールして Azure サブスクリプションに関連付けてある場合は、App Service にデプロイされたアプリに対して示されるマネージド ID の方法を使うこともできます。

Azure SQL Database に接続するコードを追加する

プロジェクト フォルダーに app.py ファイルを作成して、サンプル コードを追加します。 このコードでは、次のことを行う API が作成されます。

  • 環境変数から Azure SQL Database の接続文字列を取得します。
  • 起動時にデータベース内に Persons テーブルを作成します (テスト シナリオ専用)。
  • データベースからすべての Person レコードを取得する関数を定義します。
  • データベースからす 1 つの Person レコードを取得する関数を定義します。
  • データベースに新しい Person レコードを追加する関数を定義します。
import os
import pyodbc, struct
from azure import identity

from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel

class Person(BaseModel):
    first_name: str
    last_name: Union[str, None] = None
    
connection_string = os.environ["AZURE_SQL_CONNECTIONSTRING"]

app = FastAPI()

@app.get("/")
def root():
    print("Root of Person API")
    try:
        conn = get_conn()
        cursor = conn.cursor()

        # Table should be created ahead of time in production app.
        cursor.execute("""
            CREATE TABLE Persons (
                ID int NOT NULL PRIMARY KEY IDENTITY,
                FirstName varchar(255),
                LastName varchar(255)
            );
        """)

        conn.commit()
    except Exception as e:
        # Table may already exist
        print(e)
    return "Person API"

@app.get("/all")
def get_persons():
    rows = []
    with get_conn() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM Persons")

        for row in cursor.fetchall():
            print(row.FirstName, row.LastName)
            rows.append(f"{row.ID}, {row.FirstName}, {row.LastName}")
    return rows

@app.get("/person/{person_id}")
def get_person(person_id: int):
    with get_conn() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM Persons WHERE ID = ?", person_id)

        row = cursor.fetchone()
        return f"{row.ID}, {row.FirstName}, {row.LastName}"

@app.post("/person")
def create_person(item: Person):
    with get_conn() as conn:
        cursor = conn.cursor()
        cursor.execute(f"INSERT INTO Persons (FirstName, LastName) VALUES (?, ?)", item.first_name, item.last_name)
        conn.commit()

    return item

def get_conn():
    credential = identity.DefaultAzureCredential(exclude_interactive_browser_credential=False)
    token_bytes = credential.get_token("https://database.windows.net/.default").token.encode("UTF-16-LE")
    token_struct = struct.pack(f'<I{len(token_bytes)}s', len(token_bytes), token_bytes)
    SQL_COPT_SS_ACCESS_TOKEN = 1256  # This connection option is defined by microsoft in msodbcsql.h
    conn = pyodbc.connect(connection_string, attrs_before={SQL_COPT_SS_ACCESS_TOKEN: token_struct})
    return conn

警告

サンプル コードでは生の SQL ステートメントが示されており、運用環境のコードではこれを使わないでください。 代わりに、データベースにアクセスするためのより安全なオブジェクト レイヤーを生成する、SqlAlchemy のようなオブジェクト リレーショナル マッパー (ORM) パッケージを使ってください。

アプリをローカルで実行してテストする

アプリをローカルでテストする準備ができました。

  1. Visual Studio Code で app.py ファイルを実行します。

    uvicorn app:app --reload
    
  2. アプリ用の Swagger UI ページ http://127.0.0.1:8000/docs で、POST メソッドを展開し、[Try it out] (試してみる) を選びます。

    try /redoc を使って、API 用に生成された別の形式のドキュメントを表示することもできます。

  3. 姓と名の値を含むようにサンプル JSON を変更します。 [実行] を選択して、新しいレコードをデータベースに追加します。 API は正常な応答を返します。

  4. Swagger UI ページで GET メソッドを展開し、[テスト] を選択します。 [実行] を選択すると、先ほど作成した人物が返されます。

Azure App Service にデプロイする

アプリを Azure にデプロイする準備ができました。

  1. Azure App Service の gunicorn で uvicorn を実行できるように、start.sh ファイルを作成します。 start.sh の内容は次の 1 行です。

    gunicorn -w 4 -k uvicorn.workers.UvicornWorker app:app
    
  2. az webapp up を使って、コードを App Service にデプロイします。 (オプション -dryrun を使うと、リソースを作成せずにコマンドの動作を確認できます)。

    az webapp up \
        --resource-group <resource-group-name> \
        --name <web-app-name>         
    
  3. az webapp config set コマンドを使って、start.sh ファイルを使うように App Service を構成します。

    az webapp config set \
        --resource-group <resource-group-name> \
        --name <web-app-name> \
        --startup-file start.sh
    
  4. az webapp identity assign コマンドを使って、App Service のシステム割り当てマネージド ID を有効にします。

    az webapp identity assign \
        --resource-group <resource-group-name> \
        --name <web-app-name>
    

    このクイックスタートでは、デモのためにシステム割り当てマネージド ID を使います。 さらに広範なシナリオでは、ユーザー割り当てマネージド ID の方が効率的です。 詳しくは、「マネージド ID のベスト プラクティスに関する推奨事項」をご覧ください。 pyodbc でユーザー割り当てマネージド ID を使用する例については、「Azure SQL Database との間でパスワードレス接続を使うように Python アプリケーションを移行する」を参照してください。

App Service を Azure SQL Database に接続する

データベースを構成する」セクションでは、Azure SQL Database サーバー用のネットワークと Microsoft Entra 認証を構成しました。 このセクションでは、データベースの構成を完了し、データベース サーバーにアクセスするための接続文字列を使って App Service を構成します。

これらのコマンドを実行するには、SQL Server Management Studio (SSMS)Azure Data Studio、Visual Studio Code と SQL Server mssql 拡張機能など、Azure SQL Database に接続できる任意のツールまたは IDE を使用できます。 また、「クイック スタート: Azure portal のクエリ エディターを使用して Azure SQL Database のクエリを実行する」で説明されているように、Azure portal を使うこともできます。

  1. SQL コマンドを使ってユーザーを Azure SQL Database に追加し、パスワードレス アクセス用のユーザーとロールを作成します。

    CREATE USER [<web-app-name>] FROM EXTERNAL PROVIDER
    ALTER ROLE db_datareader ADD MEMBER [<web-app-name>]
    ALTER ROLE db_datawriter ADD MEMBER [<web-app-name>]
    

    詳細については、「 包含データベース ユーザー - データベースの可搬性を確保する」を参照してください。 同じ原則を Azure VM に適用する例については、「チュートリアル:Windows VM のシステム割り当てマネージド ID を使用して Azure SQL にアクセスする」をご覧ください。 割り当てられるロールについて詳しくは、「固定データベース ロール」をご覧ください。

    App Service のシステム割り当てマネージド ID をいったん無効にしてから有効にする場合は、ユーザーを削除して作成し直します。 DROP USER [<web-app-name>] コマンドを実行し、CREATEALTER コマンドを実行し直します。 ユーザーを表示するには、SELECT * FROM sys.database_principals を使います。

  2. az webapp config appsettings set コマンドを使って、接続文字列のためのアプリ設定を追加します。

    az webapp config appsettings set \
        --resource-group <resource-group-name> \
        --name <web-app-name> \
        --settings AZURE_SQL_CONNECTIONSTRING="<connection-string>"
    

    デプロイされたアプリの接続文字列は次のようになります。

    Driver={ODBC Driver 18 for SQL Server};Server=tcp:<database-server-name>.database.windows.net,1433;Database=<database-name>;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30
    

    <dabaser-server-name><database-name> では、実際の値を指定します。

    パスワードレスの接続文字列には、ユーザー名またはパスワードは含まれません。 代わりに、アプリが Azure で実行されると、コードは Azure ID ライブラリDefaultAzureCredential を使って、pyodbc で使うトークンを取得します。

デプロイされたアプリケーションをテストする

アプリの URL を参照して、Azure SQL Database への接続が動作していることをテストします。 アプリの URL は、App Service の概要ページで確認できます。

https://<web-app-name>.azurewebsites.net

Swagger UI を表示して API のメソッドをテストするには、/docs を URL に追加します。

おめでとうございます。 これで、ローカル環境とホスト環境の両方で、アプリケーションを Azure SQL Database に接続することができました。