Verwenden Sie mssql-python mit SQLAlchemy

SQLAlchemy ist das am weitesten verbreitete Python-ORM- und Datenbank-Toolkit. Beginnend mit SQLAlchemy 2.1.0b2 ermöglicht ein integrierter Dialekt für den mssql-python-Treiber die Verwendung von SQLAlchemy ORM und Core mit Microsoft SQL und Azure SQL-Datenbank.

Voraussetzungen

• Python 3.10 oder höher. SQLAlchemy 2.1 hat die Unterstützung für Python 3.9 und früher eingestellt.
• Die Pakete mssql-python und sqlalchemy (2.1.0b2 oder höher).

Die Beispiele in diesem Artikel verwenden die AdventureWorksLT-Beispieldatenbank . Wenn Sie AdventureWorksLT nicht installiert haben, sehen Sie sich die AdventureWorks-Beispieldatenbanken an.

Installieren Sie die Vorabversion

Da SQLAlchemy 2.1 sich in der Beta befindet, pip install sqlalchemy installiert standardmäßig die neueste stabile Version 2.0.x. Installieren Sie die Pre-Release ausdrücklich:

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

Überprüfen Sie die installierte Version:

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

Verbindungs-URLs

Der mssql-python-Dialekt verwendet mssql+mssqlpython als URL-Schema. Das allgemeine Format lautet:

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

SQL-Authentifizierung

Für die SQL-Authentifizierung fügen Sie Benutzername und Passwort in die Verbindungs-URL ein:

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

Für die Microsoft Entra-Authentifizierung verwenden Sie einen leeren Benutzernamen und den authentication Abfrageparameter:

from sqlalchemy import create_engine

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

URLs programmatisch erstellen

Verwenden Sie sqlalchemy.engine.URL.create, um eine manuelle URL-Kodierung zu vermeiden:

from sqlalchemy.engine import URL

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

Definiere ORM-Modelle

Verwenden Sie das deklarative Mapping von SQLAlchemy, um Modelle zu definieren, die auf Microsoft SQL-Tabellen abgebildet sind.

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-Vorgänge

Die folgenden Beispiele zeigen, wie man Zeilen mit der ORM-Sitzung einfügt, abfragt, aktualisiert und löscht. Jedes Beispiel verwendet new_id, das ProductID zurückgegeben wird, wenn man eine Zeile einfügt. Um alle vier Operationen gemeinsam auszuführen, siehe das vollständige Beispiel.

Erstellen Sie eine Sitzung

Erstellen Sie eine Sitzung, um Operationen innerhalb einer Transaktion auszuführen:

from sqlalchemy.orm import Session

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

Für Anwendungen, die viele Sitzungen erstellen, verwenden Sie sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Einfügen von Zeilen

Fügen Sie ein neues Produkt hinzu, schließen Sie die Sitzung ab und erfassen Sie das generierte ProductID für die folgenden Beispiele:

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

Abfragezeilen

Eine einzelne Zeile per Primärschlüssel abrufen oder für gefilterte Abfragen verwenden 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}")

Zeilen aktualisieren

Ändere ein Feld in einer bestehenden Zeile und commite:

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

Löschen von Zeilen

Entferne eine Zeile und führe einen Commit durch:

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

Vollständiges Beispiel

Die vorherigen Abschnitte zeigten jedes Stück einzeln. Dieser Abschnitt vereint sie zu einem eigenständigen Skript, das du kopieren, ausführen und erneut ausführen kannst.

Erstellen Sie eine Datei mit dem Namen crud.py, und fügen Sie den folgenden Code hinzu. Ersetzen Sie die Verbindungsdetails in create_engine durch Ihre eigenen (siehe Verbindungs-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}")

Führen Sie das Skript aus:

python crud.py

Sie sehen eine Ausgabe ähnlich der folgenden:

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

Das Skript löscht die Zeile, die es erstellt, sodass die einzigartigen Einschränkungen bei Name und ProductNumber beim erneuten Ausführen nicht betroffen sind. Jeder Durchlauf fügt eine neue Zeile ein, sodass ProductID jedes Mal zunimmt.

Kernanfragen

SQLAlchemy Core bietet eine SQL-Ausdrucks-API auf niedrigerer Ebene. Du kannst Core mit derselben Engine und denselben Tabellendefinitionen verwenden, einschließlich ORM-zugeordneter Klassen.

from sqlalchemy import text

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

Verwenden Sie Tabellenkonstrukte für die typsichere SQL-Generierung:

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

Verbindungspooling

SQLAlchemy verwaltet standardmäßig einen Verbindungspool. Passen Sie die Pool-Einstellungen für Ihre Arbeitsbelastung an:

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

Verwendung mit Web-Frameworks

SQLAlchemy wird häufig als Datenbankschicht für Flask und FastAPI verwendet. Der mssql-python-Dialekt funktioniert mit jedem Framework, das SQLAlchemy unterstützt.

Die folgenden Codeausschnitte zeigen das empfohlene Muster „eine Sitzung pro Anfrage“ für jedes Framework. Es sind veranschaulichende Fragmente, die das engine und Product Modell aus den vorherigen Abschnitten zugrunde legen, keine vollständigen Anwendungen. Für vollständige, betriebsfähige Anwendungen siehe die Artikel zur FastAPI-Integration und zur Flask-Integration .

FastAPI-Beispiel

Verwenden Sie eine Generatorabhängigkeit, um pro Anfrage eine Sitzung bereitzustellen:

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

Verwenden Sie einen Kontextmanager, um die Sitzung auf die Anfrage abzugrenzen:

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 Wanderungen

Alembic übernimmt Schema-Migrationen für SQLAlchemy-Projekte und arbeitet mit dem mssql-python-Dialekt. Alembics Autogenerierungsfunktion vergleicht deine Modelle mit der Live-Datenbank, sodass ein paar zusätzliche Schritte verhindern, dass Änderungen an Tabellen vorgeschlagen werden, die du nicht verwaltest.

Aufbau von Alembic

Installiere Alembic und initialisiere ein Migrationsverzeichnis:

pip install alembic
alembic init migrations

In alembic.inisetzen Sie die Verbindungs-URL:

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

Zeigen Sie Alembic auf Ihre Modelle

Autogenerate benötigt die Metadaten deiner Modelle. Fügen Sie die Modelle, die Alembic verwaltet, in ein importierbares Modul ein, wie zum Beispiel models.py. Da Autogenerate vorschlägt, jede Spalte, die ein Modell weglässt, wegzulassen, sollte ein Modell definiert werden, das seine Tabelle vollständig besitzt, anstatt das vereinfachte Product Modell aus diesem Artikel wiederzuverwenden:

# 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 ersetzen Sie target_metadata = None durch den folgenden Code. Es importiert deine Modelle und beschränkt die Autogenerierung auf die Schemata und Tabellen, die sie definieren:

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

Übergeben Sie include_name und include_schemas=True an context.configure in run_migrations_offline und run_migrations_online. Die Einstellung include_schemas=True erlaubt es Alembic, Tabellen in Nicht-Standardschemata zu sehen, wie zum Beispiel SalesLT:

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

Generiere und wende eine Migration an

Generiere eine Migration von deinen Modellen:

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

Alembic erkennt die neue Tabelle und schreibt ein Migrationsskript:

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

Das von upgrade() generierte Element erstellt die Tabelle, und downgrade() löscht sie:

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

Überprüfen Sie das Skript und wenden Sie dann alle ausstehenden Migrationen an:

alembic upgrade head

Unterschiede zum Pyodbc-Dialekt

Wenn du von mssql+pyodbcmigrierst, ist der mssql-python-Dialekt ähnlich, weil beide Treiber auf demselben ODBC-Framework basieren. Wichtige Unterschiede:

Bekannte Einschränkungen

Der mssql-python-Dialekt für SQLAlchemy ist in der Vorabversion. Bevor Sie sie in der Produktion verwenden, sollten Sie diese Implikationen verstehen:

• API-Änderungen: Methodensignaturen, Ausnahmetypen und Verhalten können sich vor der endgültigen stabilen Version ändern. Stecke deine SQLAlchemy-Version immer an einen bestimmten Pre-Release-Build (zum Beispiel ) sqlalchemy==2.1.0b2und teste Upgrades gründlich.

• Begrenzte Testung: Der Dialekt hat weniger Community-Tests als der stabile mssql+pyodbc Dialekt. Du könntest auf Randfälle oder fehlende Funktionen stoßen.

• Funktionslücken: Einige fortgeschrittene ORM- oder Core-Funktionen funktionieren möglicherweise nicht. Schau dir die SQLAlchemy MSSQL-Dialektdokumentation an und teste deine Anwendungsfälle, bevor du dich auf ein Projekt einlässt.

• Keine Support-Garantie: Microsoft und SQLAlchemy bieten Best-Effort-Support, aber Probleme könnten vor der stabilen Version nicht behoben sein.

Wann sollte man die Vorabveröffentlichung verwenden:

• Entwicklungs- und Testumgebungen
• Proof-of-Concept-Projekte
• Migration von mssql+pyodbc , wenn du die externe ODBC-Treiberabhängigkeit vermeiden möchtest
• Projekte, bei denen man auf API-Änderungen reagieren und Regressionstests durchführen kann

Wann sollte man die Pre-Release NICHT verwenden:

• Produktionssysteme mit strengen Stabilitätsanforderungen
• Mehrjährige Legacy-Anwendungen, bei denen Abhängigkeitsupdates selten sind
• Kritische Geschäfts-Workloads, bis SQLAlchemy 2.1 stabile GA erreicht

Für den neuesten Dialektstatus vor der Veröffentlichung und bekannte Probleme siehe das mssql-python GitHub-Repository.

Problembehandlung

"Kein Modul namens 'sqlalchemy.dialects.mssql.mssqlpython'"

Dieser Fehler bedeutet, dass deine installierte SQLAlchemy-Version den mssql-python-Dialekt nicht enthält. Überprüfen Sie, dass Sie 2.1.0b2 oder höher haben:

pip install "sqlalchemy>=2.1.0b2"

Verbindungsfehler

Wenn create_engine es gelingt, aber Abfragen fehlschlagen, überprüfen Sie, ob Ihre Verbindungsparameter direkt mit mssql-python funktionieren:

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

Wenn die direkte Verbindung funktioniert, SQLAlchemy jedoch nicht, prüfe, ob Sonderzeichen in deinem Passwort oder Servernamen Probleme bei der URL-Codierung verursachen.

Verwandte Inhalte

• mssql-python Verbindungszeichenfolgen
• MSSQL-Python Connection Management
• FastAPI-Integration
• Flask-Integration
• SQLAlchemy MSSQL-Dialekt-Dokumentation