Använd mssql-python med SQLAlchemy

SQLAlchemy är det mest använda Python-verktygslådan för ORM och databaser. SQLAlchemy 2.1 innehåller en inbyggd dialekt för drivrutinen mssql-python som du kan använda för att arbeta med SQLAlchemy ORM och Core med Microsoft SQL och Azure SQL Database.

Förutsättningar

Python 3.11 eller senare. SQLAlchemy 2.1 slutade stödja Python 3.10 och tidigare.
Paketen sqlalchemy och mssql-python (2.1 eller senare).

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

Installera SQLAlchemy och mssql-python

Använd SQLAlchemysmssql-python valfria beroende för att installera båda paketen:

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

SQLAlchemy definierar mssql-python extra med mssql-python>=1.9.0. Det befintliga separatpaketsformuläret stöds också:

pip install "sqlalchemy>=2.1" mssql-python

Kontrollera den installerade versionen:

import sqlalchemy
print(sqlalchemy.__version__)  # Should show 2.1.0 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	Description
`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:

Topic	`mssql+pyodbc`	`mssql+mssqlpython`
Installation av ODBC-drivrutin	Kräver separat ODBC-drivrutin (till exempel ODBC Driver 18 för SQL Server).	Installerat automatiskt som ett paketberoende. Ingen separat ODBC-drivrutinsinstallation 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.
Tillgänglighet	Stabilt, inkluderat i SQLAlchemy sedan version 1.x.	Stabilt, inkluderat i SQLAlchemy 2.1 eller senare.

Att tänka på när du uppgraderar

SQLAlchemy 2.1 inkluderar beteendeförändringar som kan påverka applikationer som uppgraderar från tidigare versioner:

SQLAlchemy 2.1 kräver Python 3.11 eller senare.
Uppgradera SQLAlchemy 1.x-applikationer till SQLAlchemy 2.0 innan du går över till 2.1.
Testa befintliga SQLAlchemy 2.0-applikationer mot de beteendeförändringar som beskrivs i What's New in SQLAlchemy 2.1?

Exemplen i denna artikel använder SQLAlchemys synkrona API. Applikationer som använder SQLAlchemys asyncio-stöd måste installera tillägget asyncio, eftersom SQLAlchemy 2.1 inte längre installerar greenlet som standard.

Troubleshooting

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

Detta fel innebär att din installerade SQLAlchemy-version inte inkluderar mssql-python dialekten. Installera SQLAlchemy 2.1 eller senare med dess mssql-python beroende:

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

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.

Feedback

Var den här sidan till hjälp?

Last updated on 2026-09-25

Använd mssql-python med SQLAlchemy

Förutsättningar

Installera SQLAlchemy och mssql-python

Anslutnings-URL:er

SQL-autentisering

Microsoft Entra-autentisering

Bygg URL:er programmatiskt

Definiera ORM-modeller

CRUD-operationer

Skapa en session

Infoga rader

Frågeresultatrader

Uppdatera rader

Ta bort rader

Fullständigt exempel

Kärnfrågor

Anslutningspoolning

Användning med webbramverk

FastAPI-exempel

Exempel med flaska

Alembiska migrationer

Konfigurera Alembic

Peka Alembic på dina modeller

Generera och tillämpa en migration

Skillnader från pyodbc-dialekten

Att tänka på när du uppgraderar

Troubleshooting

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

Anslutningsfel

Relaterat innehåll

Feedback

Ytterligare resurser