Gebruik mssql-python met SQLAlchemy

SQLAlchemy is de meest gebruikte Python ORM- en databasetoolkit. Te beginnen met SQLAlchemy 2.1.0b2, een ingebouwde dialect voor de mssql-python driver, waarmee je SQLAlchemy ORM en Core kunt gebruiken met Microsoft SQL en Azure SQL Database.

Prerequisites

• Python 3.10 of hoger. SQLAlchemy 2.1 stopte met de ondersteuning voor Python 3.9 en eerder.
• De pakketten mssql-python en sqlalchemy (2.1.0b2 of hoger).

De voorbeelden in dit artikel gebruiken de voorbeelddatabase AdventureWorksLT . Als je AdventureWorksLT niet hebt geïnstalleerd, zie dan de voorbeelddatabases van AdventureWorks.

Installeer de pre-release

Omdat SQLAlchemy 2.1 in bèta is, pip install sqlalchemy installeert standaard de nieuwste stabiele 2.0.x-versie. Installeer de pre-release expliciet:

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

Controleer de geïnstalleerde versie:

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

Verbindings-URL's

Het mssql-python-dialect gebruikt mssql+mssqlpython als URL-schema. De algemene indeling is:

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

SQL-verificatie

Voor SQL-authenticatie vermeld de gebruikersnaam en het wachtwoord in de verbindings-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-authenticatie

Voor Microsoft Entra-authenticatie gebruik je een lege gebruikersnaam en de authentication queryparameter:

from sqlalchemy import create_engine

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

URL's programmatisch maken

Gebruik sqlalchemy.engine.URL.create om handmatige URL-codering te vermijden:

from sqlalchemy.engine import URL

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

Definieer ORM-modellen

Gebruik de declaratieve mapping van SQLAlchemy om modellen te definiëren die worden gekoppeld aan Microsoft SQL-tabellen.

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-bewerkingen

De volgende voorbeelden laten zien hoe je rijen kunt invoegen, opvragen, bijwerken en verwijderen met behulp van de ORM-sessie. Elk voorbeeld hergebruikt new_id, de ProductID die wordt geretourneerd als je een rij invoegt. Om alle vier de bewerkingen samen uit te voeren, zie het volledige voorbeeld.

Een sessie maken

Maak een sessie aan om bewerkingen binnen een transactie uit te voeren:

from sqlalchemy.orm import Session

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

Voor applicaties die veel sessies aanmaken, gebruik sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Rijen invoegen

Voeg een nieuw product toe, commit de sessie en leg de gegenereerde ProductID gegevens vast voor de volgende voorbeelden:

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

Queryrijen

Haal een enkele rij op met primaire sleutel, of gebruik select() voor gefilterde zoekopdrachten:

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

Rijen bijwerken

Wijzig een veld op een bestaande rij en commit:

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

Rijen verwijderen

Verwijder een rij en commet:

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

Volledig voorbeeld

De voorgaande secties toonden elk stuk apart. Deze sectie combineert ze tot één zelfstandig script dat je kunt kopiëren, uitvoeren en opnieuw kunt uitvoeren.

Maak een bestand met de naam crud.py en voeg de volgende code toe. Vervang de verbindingsgegevens door create_engine je eigen (zie Verbindings-URL's):

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

Voer het script uit:

python crud.py

Je ziet een output die vergelijkbaar is met de volgende:

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

Het script verwijdert de rij die het aanmaakt, zodat het niet de unieke beperkingen NameProductNumber raakt wanneer je het opnieuw uitvoert. Elke run voegt een nieuwe rij in, dus de ProductID neemt elke keer toe.

Kernzoekopdrachten

SQLAlchemy Core biedt een SQL-expressie-API op lager niveau. Je kunt Core gebruiken met dezelfde engine- en tabeldefinities, inclusief ORM-gemapte klassen.

from sqlalchemy import text

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

Gebruik tabelniveau-constructies voor type-veilige SQL-generatie:

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

Groepsgewijze verbindingen

SQLAlchemy beheert standaard een verbindingspool. Stel de poolinstellingen aan voor je werkdruk:

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

Gebruik met webframeworks

SQLAlchemy wordt vaak gebruikt als databaselaag voor Flask en FastAPI. Het mssql-python dialect werkt met elk framework dat SQLAlchemy ondersteunt.

De volgende fragmenten tonen het aanbevolen sessie-per-verzoekpatroon voor elk framework. Het zijn illustratieve fragmenten die het engine model Product van de eerdere secties aannemen, geen volledige apps. Voor volledige, uitvoerbare applicaties, zie de artikelen over FastAPI-integratie en Flask-integratie .

FastAPI voorbeeld

Gebruik een generatorafhankelijkheid om per verzoek een sessie te leveren:

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

Fles-voorbeeld

Gebruik een contextmanager om de sessie aan het verzoek te binden:

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

Alembische migraties

Alembic verzorgt schema-migraties voor SQLAlchemy-projecten en werkt met het mssql-python-dialect. Alembic's autogenererende functie vergelijkt je modellen met de live database, dus een paar extra stappen voorkomen dat het wijzigingen voorstelt in tabellen die je niet beheert.

Alembic opzetten

Installeer Alembic en initialiseer een map voor migraties:

pip install alembic
alembic init migrations

In alembic.ini, stel de verbindings-URL in:

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

Wijs Alembic naar je modellen

Autogenerate heeft de metadata van je modellen nodig. Plaats de modellen die Alembic beheert in een importeerbare module, zoals models.py. Omdat autogenerate voorstelt om elke kolom die een model weglaat te laten vallen, definieer een model dat volledig eigenaar is van zijn tabel in plaats van het vereenvoudigde Product model van eerder in dit artikel te hergebruiken:

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

In migrations/env.py, vervang target_metadata = None door de volgende code. Het importeert je modellen en beperkt autogenereren tot de schema's en tabellen die zij definiëren:

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 en include_schemas=True naar context.configure in zowel run_migrations_offline als run_migrations_online. De include_schemas=True instelling laat Alembic tabellen zien in niet-standaard schema's zoals SalesLT:

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

Genereer en pas een migratie toe

Genereer een migratie vanuit je modellen:

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

Alembic detecteert de nieuwe tabel en schrijft een migratiescript:

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

De gegenereerde upgrade() maakt de tabel aan en downgrade() laat deze vallen:

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

Bekijk het script en pas vervolgens alle lopende migraties toe:

alembic upgrade head

Verschillen met het pyodbc-dialect

Als je migreert vanaf mssql+pyodbc, is het mssql-python-dialect vergelijkbaar, omdat beide drivers gebaseerd zijn op hetzelfde ODBC-framework. Belangrijkste verschillen:

Bekende beperkingen

Het mssql-python-dialect voor SQLAlchemy is een pre-release. Voordat je het in productie gebruikt, moet je deze implicaties begrijpen:

• API-wijzigingen: Methodehandtekeningen, uitzonderingstypen en gedrag kunnen veranderen vóór de definitieve stabiele release. Koppel je SQLAlchemy-versie altijd aan een specifieke pre-release build (bijvoorbeeld sqlalchemy==2.1.0b2) en test upgrades grondig.

• Beperkte testen: Het dialect heeft minder community-testing dan het stabiele mssql+pyodbc dialect. Je kunt randgevallen of ontbrekende functies tegenkomen.

• Functie-hiaten: Sommige geavanceerde ORM- of Core-functies werken mogelijk niet. Raadpleeg de SQLAlchemy MSSQL dialectdocumentatie en test je use cases voordat je aan een project begint.

• Geen Supportgarantie: Microsoft en SQLAlchemy bieden best effort ondersteuning, maar problemen zijn mogelijk niet opgelost vóór de stabiele release.

Wanneer de pre-release te gebruiken:

• Ontwikkel- en testomgevingen
• Proof-of-concept projecten
• Migreren van mssql+pyodbc als je de externe ODBC-driverafhankelijkheid wilt vermijden
• Projecten waar je kunt reageren op API-wijzigingen en regressietests kunt uitvoeren

Wanneer je de pre-release NIET gebruikt:

• Productiesystemen met strikte stabiliteitseisen
• Meerjarige legacy-applicaties waar afhankelijkheidsupdates zeldzaam zijn
• Kritieke bedrijfsworkloads totdat SQLAlchemy 2.1 stabiele GA bereikt

Voor de nieuwste dialectstatus en bekende problemen vóór release, raadpleeg de mssql-python GitHub-repository.

Probleemoplossingsproces

Geen module met de naam 'sqlalchemy.dialects.mssql.mssqlpython'

Deze fout betekent dat je geïnstalleerde SQLAlchemy-versie het mssql-python-dialect niet bevat. Controleer of je 2.1.0b2 of hoger hebt:

pip install "sqlalchemy>=2.1.0b2"

Verbindingsfouten

Als create_engine het lukt maar de queries mislukken, controleer dan of je verbindingsparameters direct werken met 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()

Als de directe verbinding werkt maar SQLAlchemy niet, controleer dan op URL-coderingsproblemen in speciale tekens binnen je wachtwoord of servernaam.

Verwante onderwerpen

• mssql-python verbindingsreeksen
• mssql-python verbindingsbeheer
• FastAPI-integratie
• Flask-integratie
• SQLAlchemy MSSQL dialect documentation