PythonのFastAPIとAzure SQL Databaseを接続したい

FastAPIでAPIを開発してデプロイする時、開発環境のDBはMySQL、本番環境ではAzureのSQL Databaseを利用するとします。

開発環境用（MySQL）のデータベース接続方法

1. 必要なライブラリをインストール

mysqlへの接続に利用...pip install pymsql
環境変数の読み込みに利用...pip install python-dotenv

2. 接続情報をenvファイルに記載

.env

SERVER_URL=<MySQLサーバーのURL>
DATABASE=<データベース名>
USER_NAME=<ユーザー名>
PASSWORD=<パスワード>
SERVER_PORT=<ポート番号>

3. 接続するデータベースURLを作成

config.py

import os, urllib
from dotenv import load_dotenv

load_dotenv()

SERVER_URL=os.environ["SERVER_URL"]
DATABASE=os.environ["DATABASE"]
USER_NAME=os.environ["USER_NAME"]
PASSWORD=os.environ["PASSWORD"]
SERVER_PORT=os.environ["SERVER_PORT"]

DATABASE_URL = f"mysql+pymysql://{USER_NAME}:{PASSWORD}@{SERVER_URL}:{SERVER_PORT}/{DATABASE}"

4. database.pyで読み込み

database.py

engine = create_engine(
    DATABASE_URL,echo=False, pool_pre_ping=True
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

本番環境用(Azure SQL Database)のデータベース接続方法

1. 必要なライブラリをインストール

接続にはODBC Driverを利用します。もしローカルで本番環境との接続を確認したい場合は、ローカル環境にもODBC Driverをインストールする必要があります。

Mac用

Linux用

Windows用

2. 接続情報をenvファイルに記載

こちらに記載する情報はAzure Portalから作成したSQL Databaseのリソースに行って確認してください。

.env

SERVER_URL=<SQL DatabaseのURL>
DATABASE=<データベース名>
USER_NAME=<ユーザー名>
PASSWORD=<パスワード>
DRIVER='{ODBC Driver 18 for SQL Server}' or <インストールしたバージョンのODBC>

3. 接続するデータベースURLを作成

SQL Databaseへの接続方法は何種類かありますが、今回はODBCとSQL認証を利用しています。
他にもC#やPHP、Goには専用の方法があったりしますので、Azure PortalのSQL Databaseリソース内の「接続文字列」を確認してみてください。

config.py

import os, urllib

load_dotenv()

SERVER_URL=os.environ["SERVER_URL"]
DATABASE=os.environ["DATABASE"]
USER_NAME=os.environ["USER_NAME"]
PASSWORD=os.environ["PASSWORD"]
DRIVER=os.environ["DRIVER"]

DATABASE_URL = urllib.parse.quote_plus(f'Driver={DRIVER};Server=tcp:{SERVER_URL};Database={DATABASE};Uid={USER_NAME};Pwd={PASSWORD};Encrypt=yes')
DATABASE_URL = f'mssql+pyodbc:///?odbc_connect={DATABASE_URL}'

4. database.pyで読み込み

こちらはローカル環境の時と同様です。

database.py

engine = create_engine(
    DATABASE_URL,echo=False, pool_pre_ping=True
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()