Använd mssql-python med SQLAlchemy

SQLAlchemy är det mest använda Python-verktygslådan för ORM och databaser. Med start i SQLAlchemy 2.1.0b2 låter en inbyggd dialekt för mssql-python-drivrutinen dig använda SQLAlchemy ORM och Core med Microsoft SQL och Azure SQL Database.

Important

Mssql-python-dialekten lades till i SQLAlchemy 2.1.0b2 (släpptes 16 april 2026). SQLAlchemy 2.1 är för närvarande en förhandsversion och rekommenderas inte för produktionsbruk. Innan du uppgraderar från SQLAlchemy 2.0, förstå:

• API:er kan ändras innan den slutliga stabila releasen (2.1 GA)
• Testa noggrant din arbetsbelastning innan utplacering
• Använd stabil SQLAlchemy 2.0.x för produktionssystem tills 2.1 når GA
• Fäst ditt beroende till en specifik version (till exempel sqlalchemy==2.1.0b2) istället för att använda versionsintervall

Se avsnittet Kända begränsningar för detaljer om när du ska använda förversioner.

Förutsättningar

• Python 3.10 eller senare. SQLAlchemy 2.1 slutade stödja Python 3.9 och tidigare.
• mssql-python- och sqlalchemy-paketen (2.1.0b2 eller senare).

Exemplen i denna artikel använder AdventureWorksLT :s exempeldatabas. Om du inte har AdventureWorksLT installerat, se AdventureWorks exempeldatabaser.

Installera förhandsversionen

Eftersom SQLAlchemy 2.1 är i beta, pip install sqlalchemy installerar den den senaste stabila 2.0.x-versionen som standard. Installera förhandsversionen uttryckligen:

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

Kontrollera den installerade versionen:

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

Anslutnings-URL:er

Dialekten mssql-python använder mssql+mssqlpython som URL-schema. Det allmänna formatet är:

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

SQL-autentisering

För SQL-autentisering, inkludera användarnamn och lösenord i anslutnings-URL:en:

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

För Microsoft Entra-autentisering, använd ett tomt användarnamn och frågeparameternauthentication:

from sqlalchemy import create_engine

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

Note

ActiveDirectoryDefault använder DefaultAzureCredential, som provar flera leverantörer av autentiseringsuppgifter i följd. Den första anslutningen kan vara långsam eftersom SDK:n går igenom kedjan tills den hittar en fungerande leverantör. I produktion, om du vet vilken typ av behörighet din miljö använder, ange det direkt (till exempel ActiveDirectoryMSI för managed identity) för att undvika kedjevandring. Mer information finns i Microsoft Entra-autentisering.

Bygg URL:er programmatiskt

Använd sqlalchemy.engine.URL.create för att undvika manuell URL-kodning:

from sqlalchemy.engine import URL

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

Definiera ORM-modeller

Använd SQLAlchemys deklarativa mappning för att definiera modeller som mappas till Microsoft SQL-tabeller.

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 använder IDENTITY för kolumner med automatisk inkrementering. SQLAlchemy mappar detta automatiskt för heltalskolumner i primärnyckeln. Det explicita Identity() som visas ovan är valfritt, såvida du inte behöver styra start- och inkrementvärdena.

CRUD-operationer

Följande exempel visar hur man infogar, frågar, uppdaterar och tar bort rader med hjälp av ORM-sessionen. Varje exempel återanvänder new_id, ProductID som returneras när du infogar en rad. För att köra alla fyra operationer tillsammans, se det fullständiga exemplet.

Skapa en session

Skapa en session för att utföra operationer inom en transaktion:

from sqlalchemy.orm import Session

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

För applikationer som skapar många sessioner, använd sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Infoga rader

Lägg till en ny produkt, verkställ sessionen och fånga det genererade ProductID för följande exempel:

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

I SalesLT.Producthar både Name och ProductNumber unika begränsningar. Om du kör denna insert mer än en gång, ändra dessa värden eller ta bort den tidigare raden först. Det kompletta exemplet tar bort raden det skapar, så att det kan köras upprepade gånger.

Frågeresultatrader

Hämta en enda rad med primärnyckel, eller använd select() för filtrerade frågor:

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

Uppdatera rader

Ändra ett fält på en befintlig rad och bekräfta ändringen:

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

Ta bort rader

Ta bort en rad och checka in:

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

Fullständigt exempel

De tidigare sektionerna visade varje stycke separat. Denna sektion kombinerar dem till ett självständigt skript som du kan kopiera, köra och köra igen.

Skapa en fil med namnet crud.py och lägg till följande kod. Byt ut anslutningsdetaljerna mot create_engine dina egna (se Anslutnings-URL:er):

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

Kör skriptet:

python crud.py

Du ser ett resultat liknande följande:

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

Skriptet raderar raden det skapar, så det träffar inte de unika begränsningarna på Name och ProductNumber när du kör det igen. Varje körning infogar en ny rad, så ProductID ökar varje gång.

Kärnfrågor

SQLAlchemy Core tillhandahåller ett API för SQL-uttryck på lägre nivå. Du kan använda Core med samma motor och tabelldefinitioner, inklusive ORM-mappade klasser.

from sqlalchemy import text

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

Använd tabellnivåkonstruktioner för typsäker SQL-generering:

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

När du väljer individuella mappade kolumner vars databasnamn skiljer sig från attributnamnet (till exempel Product.name mappar till kolumnen Name ), är Core-raderna nyckelade med databasens kolumnnamn. Lägg till .label("name") för att komma åt värdet som row.name istället för row.Name.

Anslutningspoolning

SQLAlchemy hanterar en anslutningspool som standard. Justera poolinställningarna för din arbetsbelastning:

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

Parameter	Beskrivning
pool_size	Antal anslutningar att hålla öppna (standard: 5).
max_overflow	Anslutningar tillåtna bortom pool_size (standard: 10).
pool_timeout	Sekunder att vänta på att en anslutning upprättas innan ett fel genereras (standardvärde: 30).
pool_recycle	Sekunder därefter återvinns en anslutning (standard: -1, inaktiverad). Sätt detta värde om din databas stänger inaktiva anslutningar.

Användning med webbramverk

SQLAlchemy används ofta som databaslager för Flask och FastAPI. mssql-python-dialekten fungerar med alla ramverk som stödjer SQLAlchemy.

Följande utdrag visar det rekommenderade session-per-förfrågan-mönstret för varje ramverk. De är illustrativa fragment som utgår från modellen engine och Product från de tidigare avsnitten, inte fullständiga appar. För kompletta, körbara applikationer, se artiklarna om FastAPI-integration och Flask-integration .

FastAPI-exempel

Använd ett generatorberoende för att tillhandahålla en session per förfrågan:

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

Exempel med flaska

Använd en kontexthanterare för att begränsa sessionen till begäran:

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

Alembiska migrationer

Alembic hanterar schemamigreringar för SQLAlchemy-projekt och fungerar med mssql-python-dialekten. Alembics autogenereringsfunktion jämför dina modeller med den levande databasen, så några extra steg hindrar den från att föreslå ändringar i tabeller du inte hanterar.

Konfigurera Alembic

Installera Alembic och initiera en migrationskatalog:

pip install alembic
alembic init migrations

I alembic.ini, ställ anslutnings-URL:n:

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

Peka Alembic på dina modeller

Autogenerate behöver dina modellers metadata. Lägg modellerna som Alembic hanterar i en importerbar modul, såsom models.py. Eftersom autogenerate föreslår att man tar bort vilken kolumn som helst som en modell utelämnar, definiera en modell som helt äger sin tabell istället för att återanvända den förenklade Product modellen från tidigare i denna artikel:

# 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

Som standard behandlar autogenerate varje tabell i databasen som inte är i target_metadata som borttagen och skickar drop_table ut för den. Mot en befintlig databas som AdventureWorksLT kan den åtgärden släppa dussintals tabeller. Lägg till ett include_name filter så att Alembic bara hanterar de tabeller som dina modeller definierar, och granska alltid det genererade skriptet innan du applicerar det.

I migrations/env.py, ersätt target_metadata = None med följande kod. Den importerar dina modeller och begränsar autogenerering till de scheman och tabeller de definierar:

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 och include_schemas=True till context.configure i både run_migrations_offline och run_migrations_online. Inställningen include_schemas=True låter Alembic se tabeller i icke-standardscheman såsom SalesLT:

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

Generera och tillämpa en migration

Generera en migration från dina modeller:

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

Alembic upptäcker den nya tabellen och skriver ett migrationsskript:

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

Den genererade upgrade() skapar tabellen och downgrade() släpper den:

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

Gå igenom skriptet och tillämpa sedan alla väntande migreringar:

alembic upgrade head

Skillnader från pyodbc-dialekten

Om du migrerar från mssql+pyodbc, är mssql-python-dialekten liknande eftersom båda drivrutinerna bygger på samma ODBC-ramverk. Viktiga skillnader:

Ämne	mssql+pyodbc	mssql+mssqlpython
Installation av ODBC-drivrutin	Kräver separat ODBC-drivrutin (till exempel ODBC Driver 18 för Microsoft SQL).	Drivrutinen medföljer. Ingen separat ODBC-drivrutin behövs.
Anslutnings-URL	mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server	mssql+mssqlpython://user:pass@host/db
fast_executemany	Stödd via create_engine(..., fast_executemany=True).	Ej tillämpbart. Drivrutinen hanterar batchprestanda internt.
Availability	Stabilt, inkluderat i SQLAlchemy sedan version 1.x.	Förhandsversion (SQLAlchemy 2.1.0b2+).

Kända begränsningar

MSSQL-python-dialekten för SQLAlchemy finns i förhandsversion. Innan du använder i produktion, förstå dessa konsekvenser:

• API-ändringar: Metodsignaturer, undantagstyper och beteende kan ändras innan den slutliga stabila releasen. Fäst alltid din SQLAlchemy-version vid en specifik förhandsversion (till exempel sqlalchemy==2.1.0b2) och testa uppgraderingar noggrant.

• Begränsad testning: Dialekten har mindre community-testning än den stabila mssql+pyodbc dialekten. Du kan stöta på undantagsfall eller saknade funktioner.

• Funktionsluckor: Vissa avancerade ORM- eller Core-funktioner kanske inte fungerar. Se SQLAlchemy MSSQL-dialektdokumentation och testa dina användningsfall innan du bestämmer dig för ett projekt.

• Ingen supportgaranti: Microsoft och SQLAlchemy erbjuder bästa möjliga support, men problemen kanske inte löses innan den stabila releasen.

När ska man använda förhandsversionen:

• Utvecklings- och testmiljöer
• Proof-of-concept-projekt
• Migrerar från mssql+pyodbc om du vill undvika extern ODBC-drivrutinsberoende
• Projekt där du kan svara på API-ändringar och utföra regressionstester

När man INTE ska använda förhandsversionen:

• Produktionssystem med strikta stabilitetskrav
• Fleråriga äldre applikationer där beroendeuppdateringar är sällsynta
• Kritiska affärsarbetsbelastningar tills SQLAlchemy 2.1 når stabil GA

För senaste dialektstatus och kända problem före release, kolla mssql-python GitHub-arkivet.

Troubleshooting

"Ingen modul som heter 'sqlalchemy.dialects.mssql.mssqlpython'"

Detta fel innebär att din installerade SQLAlchemy-version inte inkluderar mssql-python-dialekten. Verifiera att du har 2.1.0b2 eller senare:

pip install "sqlalchemy>=2.1.0b2"

Anslutningsfel

Om create_engine lyckas men frågorna misslyckas, verifiera att dina anslutningsparametrar fungerar direkt med 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()

Om den direkta anslutningen fungerar men SQLAlchemy inte gör det, kontrollera om URL-kodningsproblem finns i specialtecken i ditt lösenord eller servernamn.

Relaterat innehåll

• MSSQL-Python-anslutningssträngar
• mssql-python anslutningshantering
• FastAPI-integration
• Flaskintegration
• SQLAlchemy MSSQL dialect documentation