Use mssql-python with SQLAlchemy

SQLAlchemy is the most widely used Python ORM and database toolkit. SQLAlchemy 2.1 includes a built-in dialect for the mssql-python driver that you can use to work with SQLAlchemy ORM and Core with Microsoft SQL and Azure SQL Database.

Prerequisites

Python 3.11 or later. SQLAlchemy 2.1 dropped support for Python 3.10 and earlier.
The mssql-python and sqlalchemy packages (2.1 or later).

The examples in this article use the AdventureWorksLT sample database. If you don't have AdventureWorksLT installed, see AdventureWorks sample databases.

Install SQLAlchemy and mssql-python

Use SQLAlchemy's mssql-python optional dependency to install both packages:

pip install "sqlalchemy[mssql-python]>=2.1"

SQLAlchemy defines the mssql-python extra with mssql-python>=1.9.0. The existing separate-package form is also supported:

pip install "sqlalchemy>=2.1" mssql-python

Verify the installed version:

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

Connection URLs

The mssql-python dialect uses mssql+mssqlpython as the URL scheme. The general format is:

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

SQL authentication

For SQL authentication, include the username and password in the connection 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 authentication

For Microsoft Entra authentication, use an empty username and the authentication query parameter:

from sqlalchemy import create_engine

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

Note

ActiveDirectoryDefault uses DefaultAzureCredential, which tries multiple credential providers in sequence. The first connection can be slow because the SDK walks the chain until it finds a working provider. In production, if you know which credential type your environment uses, specify it directly (for example, ActiveDirectoryMSI for managed identity) to avoid the chain walk. For more information, see Microsoft Entra authentication.

Build URLs programmatically

Use sqlalchemy.engine.URL.create to avoid manual URL encoding:

from sqlalchemy.engine import URL

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

Define ORM models

Use SQLAlchemy's declarative mapping to define models that map to Microsoft SQL tables.

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()
    )

Tip

Microsoft SQL uses IDENTITY for auto-incrementing columns. SQLAlchemy maps this automatically for integer primary key columns. The explicit Identity() shown above is optional unless you need to control the start and increment values.

CRUD operations

The following examples show how to insert, query, update, and delete rows by using the ORM session. Each example reuses new_id, the ProductID returned when you insert a row. To run all four operations together, see the complete example.

Create a session

Create a session to execute operations within a transaction:

from sqlalchemy.orm import Session

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

For applications that create many sessions, use sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Insert rows

Add a new product, commit the session, and capture the generated ProductID for the following examples:

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}")

Note

In SalesLT.Product, both Name and ProductNumber have unique constraints. If you run this insert more than once, change these values or delete the earlier row first. The complete example deletes the row it creates, so it can run repeatedly.

Query rows

Retrieve a single row by primary key, or use select() for filtered queries:

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}")

Update rows

Modify a field on an existing row and commit:

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

Delete rows

Remove a row and commit:

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

Complete example

The previous sections showed each piece separately. This section combines them into one self-contained script that you can copy, run, and run again.

Create a file named crud.py and add the following code. Replace the connection details in create_engine with your own (see Connection URLs):

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}")

Run the script:

python crud.py

You see output similar to the following:

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

The script deletes the row it creates, so it doesn't hit the unique constraints on Name and ProductNumber when you run it again. Each run inserts a new row, so the ProductID increases each time.

Core queries

SQLAlchemy Core provides a lower-level SQL expression API. You can use Core with the same engine and table definitions, including ORM-mapped classes.

from sqlalchemy import text

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

Use table-level constructs for type-safe SQL generation:

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()

Note

When you select individual mapped columns whose database name differs from the attribute name (for example, Product.name maps to the Name column), Core rows are keyed by the database column name. Add .label("name") to access the value as row.name instead of row.Name.

Connection pooling

SQLAlchemy manages a connection pool by default. Tune pool settings for your workload:

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

Parameter	Description
`pool_size`	Number of connections to keep open (default: 5).
`max_overflow`	Connections allowed beyond `pool_size` (default: 10).
`pool_timeout`	Seconds to wait for a connection before raising an error (default: 30).
`pool_recycle`	Seconds after which a connection is recycled (default: -1, disabled). Set this value if your database closes idle connections.

Use with web frameworks

SQLAlchemy is commonly used as the database layer for Flask and FastAPI. The mssql-python dialect works with any framework that supports SQLAlchemy.

The following snippets show the recommended session-per-request pattern for each framework. They're illustrative fragments that assume the engine and Product model from the earlier sections, not complete apps. For complete, runnable applications, see the FastAPI integration and Flask integration articles.

FastAPI example

Use a generator dependency to provide a session per request:

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 example

Use a context manager to scope the session to the request:

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 migrations

Alembic handles schema migrations for SQLAlchemy projects, and it works with the mssql-python dialect. Alembic's autogenerate feature compares your models to the live database, so a few extra steps keep it from proposing changes to tables you don't manage.

Set up Alembic

Install Alembic and initialize a migrations directory:

pip install alembic
alembic init migrations

In alembic.ini, set the connection URL:

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

Point Alembic at your models

Autogenerate needs your models' metadata. Put the models Alembic manages in an importable module, such as models.py. Because autogenerate proposes dropping any column that a model omits, define a model that fully owns its table rather than reusing the simplified Product model from earlier in this article:

# 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())

Caution

By default, autogenerate treats every table in the database that isn't in target_metadata as removed and emits drop_table for it. Against an existing database like AdventureWorksLT, that action can drop dozens of tables. Add an include_name filter so Alembic manages only the tables your models define, and always review the generated script before you apply it.

In migrations/env.py, replace target_metadata = None with the following code. It imports your models and limits autogenerate to the schemas and tables they define:

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

Pass include_name and include_schemas=True to context.configure in both run_migrations_offline and run_migrations_online. The include_schemas=True setting lets Alembic see tables in nondefault schemas such as SalesLT:

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

Generate and apply a migration

Generate a migration from your models:

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

Alembic detects the new table and writes a migration script:

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

The generated upgrade() creates the table, and downgrade() drops it:

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")

Review the script, and then apply all pending migrations:

alembic upgrade head

Differences from pyodbc dialect

If you're migrating from mssql+pyodbc, the mssql-python dialect is similar because both drivers are based on the same ODBC framework. Key differences:

Topic	`mssql+pyodbc`	`mssql+mssqlpython`
ODBC driver installation	Requires separate ODBC driver (for example, ODBC Driver 18 for SQL Server).	Installed automatically as a package dependency. No separate ODBC driver installation needed.
Connection URL	`mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server`	`mssql+mssqlpython://user:pass@host/db`
`fast_executemany`	Supported via `create_engine(..., fast_executemany=True)`.	Not applicable. The driver handles batch performance internally.
Availability	Stable, included in SQLAlchemy since 1.x.	Stable, included in SQLAlchemy 2.1 or later.

Upgrade considerations

SQLAlchemy 2.1 includes behavioral changes that might affect applications upgrading from earlier releases:

SQLAlchemy 2.1 requires Python 3.11 or later.
Upgrade SQLAlchemy 1.x applications to SQLAlchemy 2.0 before moving to 2.1.
Test existing SQLAlchemy 2.0 applications against the behavioral changes described in What's New in SQLAlchemy 2.1?

The examples in this article use SQLAlchemy's synchronous API. Applications that use SQLAlchemy asyncio support must install the asyncio extra because SQLAlchemy 2.1 no longer installs greenlet by default.

Troubleshooting

"No module named 'sqlalchemy.dialects.mssql.mssqlpython'"

This error means that your installed SQLAlchemy version doesn't include the mssql-python dialect. Install SQLAlchemy 2.1 or later with its mssql-python dependency:

pip install --upgrade "sqlalchemy[mssql-python]>=2.1"

Connection failures

If create_engine succeeds but queries fail, verify your connection parameters work with mssql-python directly:

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()

If the direct connection works but SQLAlchemy doesn't, check for URL encoding issues in special characters within your password or server name.

משוב

האם עמוד זה היה מועיל?

Last updated on 2026-09-25

Use mssql-python with SQLAlchemy

Prerequisites

Install SQLAlchemy and mssql-python

Connection URLs

SQL authentication

Microsoft Entra authentication

Build URLs programmatically

Define ORM models

CRUD operations

Create a session

Insert rows

Query rows

Update rows

Delete rows

Complete example

Core queries

Connection pooling

Use with web frameworks

FastAPI example

Flask example

Alembic migrations

Set up Alembic

Point Alembic at your models

Generate and apply a migration

Differences from pyodbc dialect

Upgrade considerations

Troubleshooting

"No module named 'sqlalchemy.dialects.mssql.mssqlpython'"

Connection failures

Related content

משוב

משאבים נוספים