MSSQL-pythonとSQLAlchemyを組み合わせて使う

SQLAlchemyは最も広く使われているPython ORMおよびデータベースツールキットです。 SQLAlchemy 2.1.0b2からは、mssql-pythonドライバー用の組み込み方言により、Microsoft SQLやAzure SQL DatabaseでSQLAlchemy ORMとCoreを利用できます。

前提条件

• Python 3.10 以降。 SQLAlchemy 2.1はPython 3.9以前のバージョンのサポートを終了しました。
• mssql-pythonパッケージとsqlalchemyパッケージ(2.1.0b2以降)。

この記事の例は AdventureWorksLT のサンプルデータベースを使用しています。 AdventureWorksLTをインストールしていない場合は、 AdventureWorksのサンプルデータベースをご覧ください。

プレリリースをインストールしてください

SQLAlchemy 2.1はベータ版であるため、pip install sqlalchemyは最新の安定した2.0.xリリースをデフォルトでインストールします。 プレリリースを明示的にインストールする:

pip install mssql-python "sqlalchemy>=2.1.0b2"

インストールされているバージョンを確認します。

import sqlalchemy
print(sqlalchemy.__version__)  # Should show 2.1.0b2 or later

接続URL

mssql-python方言ではURLスキームとして mssql+mssqlpython を使用します。 一般的な形式は次のとおりです。

mssql+mssqlpython://<username>:<password>@<host>:<port>/<database>

SQL 認証

SQL認証の場合、接続URLにユーザー名とパスワードを含めてください:

from sqlalchemy import create_engine

# Replace <password> with your actual password. Avoid using the sa account in production.
engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost:1433/<database>"
)

Microsoft Entra 認証

Microsoft Entra認証には空のユーザー名とauthenticationクエリパラメータを使用します。

from sqlalchemy import create_engine

engine = create_engine(
    "mssql+mssqlpython://@<server>.database.windows.net/<database>"
    "?authentication=ActiveDirectoryDefault&encrypt=yes"
)

プログラム的にURLを構築する

手動のURLエンコーディングを避けるために sqlalchemy.engine.URL.create を使います:

from sqlalchemy.engine import URL

url = URL.create(
    "mssql+mssqlpython",
    username="dbuser",
    password="<password>",
    host="localhost",
    port=1433,
    database="<database>",
)
engine = create_engine(url)

ORMモデルの定義

SQLAlchemyの宣言的マッピングを使って、MicrosoftのSQLテーブルにマッピングするモデルを定義してください。

from datetime import datetime
from decimal import Decimal

from sqlalchemy import Identity, String, Numeric, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "Product"
    __table_args__ = {"schema": "SalesLT"}

    product_id: Mapped[int] = mapped_column(
        "ProductID", Integer, Identity(), primary_key=True
    )
    name: Mapped[str] = mapped_column("Name", String(50))
    product_number: Mapped[str] = mapped_column("ProductNumber", String(25))
    color: Mapped[str | None] = mapped_column("Color", String(15))
    list_price: Mapped[Decimal] = mapped_column("ListPrice", Numeric(19, 4))
    standard_cost: Mapped[Decimal] = mapped_column("StandardCost", Numeric(19, 4))
    size: Mapped[str | None] = mapped_column("Size", String(5))
    product_category_id: Mapped[int | None] = mapped_column("ProductCategoryID", Integer)
    sell_start_date: Mapped[datetime] = mapped_column("SellStartDate", DateTime)
    modified_date: Mapped[datetime] = mapped_column(
        "ModifiedDate", DateTime, server_default=func.getdate()
    )

CRUD操作

以下の例は、ORMセッションを使って行を挿入、クエリ、更新、削除する方法を示しています。 各例では、行を挿入したときに返される ProductID である new_id を再利用します。 4つの操作すべてを同時に実行するには、 完全な例を参照してください。

セッションを作成する

トランザクション内で操作を実行するためのセッションを作成する:

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Use session for queries and modifications
    pass

多数のセッションを作成するアプリケーションには、 sessionmakerを使用します。

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

行を挿入する

新しいプロダクトを追加し、セッションをコミットし、以下の例で生成された ProductID を取得します。

from datetime import datetime

with Session(engine) as session:
    product = Product(
        name="Classic Road Bike",
        product_number="BK-C001",
        color="Red",
        list_price=Decimal("1299.99"),
        standard_cost=Decimal("749.99"),
        sell_start_date=datetime(2026, 1, 1),
        product_category_id=6,
    )
    session.add(product)
    session.commit()

    new_id = product.product_id
    print(f"Inserted ProductID: {new_id}")

クエリ行

主キーで単行を取得するか、フィルター付きクエリには select() を使います:

from sqlalchemy import select

with Session(engine) as session:
    # Single row by primary key (new_id is from the insert example)
    product = session.get(Product, new_id)
    if product:
        print(f"{product.name}: ${product.list_price}")

    # Filtered query
    stmt = select(Product).where(Product.list_price < 500).order_by(Product.name)
    products = session.scalars(stmt).all()
    for p in products:
        print(f"{p.name}: ${p.list_price}")

行の更新

既存の行のフィールドを修正し、コミットします:

with Session(engine) as session:
    product = session.get(Product, new_id)
    if product:
        product.list_price = Decimal("1349.99")
        session.commit()

行を削除する

行を削除してコミットします:

with Session(engine) as session:
    product = session.get(Product, new_id)
    if product:
        session.delete(product)
        session.commit()

完全なコード例

前のセクションでは各作品を個別に紹介しました。 このセクションではそれらを一つの独立したスクリプトにまとめ、コピーして実行し、再度実行できます。

crud.pyという名前のファイルを作成し、次のコードを追加します。 create_engineの接続情報は自分のものに置き換えてください(接続URLを参照):

from datetime import datetime
from decimal import Decimal

from sqlalchemy import create_engine, Identity, String, Numeric, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# Replace <password> and <database> with your connection details.
engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost:1433/<database>"
)


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "Product"
    __table_args__ = {"schema": "SalesLT"}

    product_id: Mapped[int] = mapped_column(
        "ProductID", Integer, Identity(), primary_key=True
    )
    name: Mapped[str] = mapped_column("Name", String(50))
    product_number: Mapped[str] = mapped_column("ProductNumber", String(25))
    color: Mapped[str | None] = mapped_column("Color", String(15))
    list_price: Mapped[Decimal] = mapped_column("ListPrice", Numeric(19, 4))
    standard_cost: Mapped[Decimal] = mapped_column("StandardCost", Numeric(19, 4))
    size: Mapped[str | None] = mapped_column("Size", String(5))
    product_category_id: Mapped[int | None] = mapped_column("ProductCategoryID", Integer)
    sell_start_date: Mapped[datetime] = mapped_column("SellStartDate", DateTime)
    modified_date: Mapped[datetime] = mapped_column(
        "ModifiedDate", DateTime, server_default=func.getdate()
    )


with Session(engine) as session:
    # Create
    product = Product(
        name="Classic Road Bike",
        product_number="BK-C001",
        color="Red",
        list_price=Decimal("1299.99"),
        standard_cost=Decimal("749.99"),
        sell_start_date=datetime(2026, 1, 1),
        product_category_id=6,
    )
    session.add(product)
    session.commit()
    new_id = product.product_id
    print(f"Inserted ProductID: {new_id}")

    # Read
    product = session.get(Product, new_id)
    print(f"Read: {product.name} costs ${product.list_price}")

    # Update
    product.list_price = Decimal("1349.99")
    session.commit()
    print(f"Updated price to ${product.list_price}")

    # Delete
    session.delete(product)
    session.commit()
    print(f"Deleted ProductID: {new_id}")

スクリプトを実行します。

python crud.py

以下のような出力が見えます:

Inserted ProductID: 1019
Read: Classic Road Bike costs $1299.9900
Updated price to $1349.9900
Deleted ProductID: 1019

スクリプトは作成した行を削除し、再度実行した際に Name と ProductNumber の固有の制約に当たらない状態です。 各ランごとに新しい行が挿入されるため、その ProductID は毎回増加します。

主要なクエリ

SQLAlchemy Coreは低レベルのSQL式APIを提供しています。 同じエンジンやテーブル定義、ORMマッピングされたクラスも含めてCoreを利用できます。

from sqlalchemy import text

with engine.connect() as conn:
    result = conn.execute(text("SELECT @@VERSION"))
    print(result.scalar())

型安全なSQL生成にはテーブルレベルの構造を使う:

from sqlalchemy import insert, select, update, delete

with engine.connect() as conn:
    # Insert
    conn.execute(
        insert(Product).values(
            name="Touring Bike",
            product_number="BK-T002",
            list_price=Decimal("999.99"),
            standard_cost=Decimal("575.00"),
            sell_start_date=datetime(2026, 1, 1)
        )
    )
    conn.commit()

    # Select
    stmt = select(
        Product.name.label("name"),
        Product.list_price.label("list_price"),
    ).where(Product.list_price > 100)
    for row in conn.execute(stmt):
        print(row.name, row.list_price)

    # Delete the inserted row so this example can run again
    conn.execute(delete(Product).where(Product.product_number == "BK-T002"))
    conn.commit()

コネクションプーリング

SQLAlchemy はデフォルトで接続プールを管理しています。 作業量に合わせてプール設定を調整する:

engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost/<database>",
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=3600,
)

ウェブフレームワークでの使用

SQLAlchemy はFlaskやFastAPIのデータベース層として一般的に使われています。 mssql-python方言は 、SQLAlchemyをサポートするあらゆるフレームワークで動作します。

以下のスニペットは、各フレームワークごとに推奨されるリクエストごとのセッションパターンを示しています。 これらは、以前のセクションの engine や Product モデルを前提とした説明的な断片であり、完全なアプリではありません。 完全で実行可能なアプリケーションについては、 FastAPI統合 および Flask統合 の記事をご覧ください。

FastAPIの例

ジェネレーター依存性を使ってリクエストごとにセッションを提供する:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

engine = create_engine("mssql+mssqlpython://dbuser:<password>@<server>/<database>")
SessionLocal = sessionmaker(bind=engine)

app = FastAPI()


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


@app.get("/products/{product_id}")
def read_product(product_id: int, db: Session = Depends(get_db)):
    product = db.get(Product, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return {"name": product.name, "price": float(product.list_price)}

Flask の例

コンテキストマネージャーを使って、セッションをリクエストにスコープします:

from flask import Flask, jsonify
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

engine = create_engine("mssql+mssqlpython://dbuser:<password>@<server>/<database>")
SessionLocal = sessionmaker(bind=engine)

app = Flask(__name__)


@app.route("/products/<int:product_id>")
def read_product(product_id):
    with SessionLocal() as session:
        product = session.get(Product, product_id)
        if not product:
            return jsonify({"error": "Not found"}), 404
        return jsonify({"name": product.name, "price": float(product.list_price)})

アレンビックの移動

Alembic は SQLAlchemy プロジェクトのスキーマ移行を担当し、mssql-python方言に対応しています。 Alembicの自動生成機能はモデルを実際のデータベースと比較するため、いくつか追加の手順を踏むことで、管理対象外のテーブルへの変更を提案しないようにできます。

アレンビックの設立

Alembicをインストールし、マイグレーションディレクトリを初期化します:

pip install alembic
alembic init migrations

alembic.iniで接続URLを設定します:

sqlalchemy.url = mssql+mssqlpython://dbuser:<password>@localhost/<database>

アレンビックをモデルに向けてください

自動生成はモデルのメタデータが必要です。 Alembicが管理するモデルをインポート可能なモジュール(例えば models.py)に入れます。 autogenerateはモデルが省略した列を除外することを提案しているため、本記事の前半で簡略化された Product モデルを再利用するのではなく、そのテーブルを完全に所有するモデルを定義してください。

# models.py
from datetime import datetime

from sqlalchemy import Identity, String, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class ProductReview(Base):
    __tablename__ = "ProductReview"
    __table_args__ = {"schema": "SalesLT"}

    review_id: Mapped[int] = mapped_column("ReviewID", Integer, Identity(), primary_key=True)
    product_id: Mapped[int] = mapped_column("ProductID", Integer)
    reviewer_name: Mapped[str] = mapped_column("ReviewerName", String(50))
    rating: Mapped[int] = mapped_column("Rating", Integer)
    comments: Mapped[str | None] = mapped_column("Comments", String(500))
    modified_date: Mapped[datetime] = mapped_column("ModifiedDate", DateTime, server_default=func.getdate())

migrations/env.pyでは、target_metadata = Noneを次のコードに置き換えます。 モデルをインポートし、オート生成を定義したスキーマやテーブルに制限します:

from models import Base

target_metadata = Base.metadata

# Limit autogenerate to the tables your models define.
managed_schemas = {table.schema for table in target_metadata.tables.values()}
managed_tables = {table.name for table in target_metadata.tables.values()}


def include_name(name, type_, parent_names):
    if type_ == "schema":
        return name in managed_schemas
    if type_ == "table":
        return name in managed_tables
    return True

run_migrations_offline と run_migrations_online の両方で、include_name と include_schemas=True を context.configure に渡してください。 include_schemas=True設定により、AlembicはSalesLTのような非デフォルトスキーマのテーブルを閲覧できます。

context.configure(
    connection=connection,
    target_metadata=target_metadata,
    include_name=include_name,
    include_schemas=True,
)

移行を生成し適用する

モデルに基づいてマイグレーションを生成する:

alembic revision --autogenerate -m "add product review table"

Alembicは新しいテーブルを検出し、マイグレーションスクリプトを書きます:

INFO  [alembic.autogenerate.compare.tables] Detected added table 'SalesLT.ProductReview'
Generating .../versions/xxxx_add_product_review_table.py ... done

生成された upgrade() がテーブルを作成し、 downgrade() それをドロップします:

def upgrade() -> None:
    op.create_table(
        "ProductReview",
        sa.Column("ReviewID", sa.Integer(), sa.Identity(always=False), nullable=False),
        sa.Column("ProductID", sa.Integer(), nullable=False),
        sa.Column("ReviewerName", sa.String(length=50), nullable=False),
        sa.Column("Rating", sa.Integer(), nullable=False),
        sa.Column("Comments", sa.String(length=500), nullable=True),
        sa.Column("ModifiedDate", sa.DateTime(), server_default=sa.text("getdate()"), nullable=False),
        sa.PrimaryKeyConstraint("ReviewID"),
        schema="SalesLT",
    )


def downgrade() -> None:
    op.drop_table("ProductReview", schema="SalesLT")

スクリプトを確認し、保留中のすべての移行を適用してください:

alembic upgrade head

ピョドブク方言との違い

mssql+pyodbcからの移行なら、mssqlとpythonの方言は似ています。両ドライバーは同じODBCフレームワークに基づいているからです。 主な違い:

既知の制限

SQLAlchemyのmssql-python方言はまだプレリリース段階です。 本番環境で使用する前に、以下の意味を理解してください:

• APIの変更:最終安定リリース前にメソッド署名、例外タイプ、挙動が変更される可能性があります。 必ず SQLAlchemy のバージョンを特定のプレリリースビルド(例: sqlalchemy==2.1.0b2)にピンインし、アップグレードを徹底的にテストしてください。

• 限定的なテスト:この方言は安定した mssql+pyodbc 方言よりもコミュニティテストが少ないです。 例外や欠落した機能に遭遇するかもしれません。

• 機能ギャップ:一部の高度なORMやコア機能が動作しない場合があります。 プロジェクトにコミットする前に、 SQLAlchemy MSSQL方言のドキュメント を参照し、ユースケースをテストしてください。

• サポート保証なし:MicrosoftとSQLAlchemyはベストエフォートサポートを提供していますが、安定版リリース前に問題が解決されない場合があります。

プレリリースの利用時期:

• 開発環境とテスト環境
• 概念実証プロジェクト
• 外部ODBCドライバー依存を避けたいなら mssql+pyodbc からの移行を検討します
• APIの変更に対応し、回帰テストを行うプロジェクト

プレリリースを使わないべき場合:

• 厳格な安定性要件を持つ生産システム
• 依存関係の更新が稀な複数年にわたるレガシーアプリケーション
• SQLAlchemy 2.1が安定したGAに到達するまでの重要なビジネスワークロード

最新のプレリリース方言状況や既知の問題については、mssql-pythonのGitHubリポジトリをご覧ください。

Troubleshooting

「『sqlalchemy.dialects.mssql.mssqlpython』という名前のモジュールは存在しません」

このエラーは、インストールされている SQLAlchemy バージョンにmssql-python方言が含まれていないことを意味します。 2.1.0b2以降のバージョンを使用していることを確認してください:

pip install "sqlalchemy>=2.1.0b2"

接続の失敗

create_engine成功してもクエリが失敗した場合は、mssql-pythonで接続パラメータが動作しているか直接確認してください:

import mssql_python

conn = mssql_python.connect(
    "Server=localhost;Database=<database>;UID=dbuser;PWD=<password>;Encrypt=yes"
)
cursor = conn.cursor()
cursor.execute("SELECT 1")
print(cursor.fetchone())
conn.close()

直接接続は動作しても SQLAlchemy が動かない場合は、パスワードやサーバー名内の特殊文字でURLエンコードの問題がないか確認してください。

関連するコンテンツ

• MSSQL-python 接続文字列
• MSSQL-python 接続管理
• FastAPI統合
• Flask 連携
• SQLAlchemy MSSQL 方言ドキュメント