Az mssql-python használata SQLAlchemyvel

Az SQLAlchemy a legelterjedtebb Python ORM és adatbázis eszköztár. Az SQLAlchemy 2.1.0b2-vel kezdve, egy beépített dialektus az mssql-python driverhez lehetővé teszi SQLAlchemy ORM és Core használatát Microsoft SQL és Azure SQL Database segítségével.

Important

Az mssql-python dialektust az SQLAlchemy 2.1.0b2-ben adták hozzá (2026. április 16-án jelent meg). Az SQLAlchemy 2.1 jelenleg előzetes kiadású sorozat, és nem ajánlott gyártási használatra. Mielőtt frissítené az SQLAlchemy 2.0-ról, értsd meg:

  • Az API-k megváltozhatnak a végleges stabil kiadás (2.1 GA) előtt
  • Teszteld alaposan a munkaterhelésedet a telepítés előtt
  • Használja a stabil SQLAlchemy 2.0.x verziót éles rendszerekben, amíg a 2.1 általánosan elérhetővé nem válik
  • Rögzítsd a függőségedet egy adott verzióhoz (például sqlalchemy==2.1.0b2) a verziótartományok helyett

Lásd az Ismert Korlátozások szekciót a részletekért arról, mikor érdemes használni az előzetes kiadású verziókat.

Prerequisites

  • Python 3.10 vagy újabb verzió. Az SQLAlchemy 2.1 megszüntette a Python 3.9 és korábbi verziók támogatását.
  • Az mssql-python and sqlalchemy csomagok (2.1.0b2 vagy újabb).

A cikkben szereplő példák az AdventureWorksLT mintaadatbázist használják. Ha nincs telepítve az AdventureWorksLT, nézd meg az AdventureWorks mintaadatbázisokat.

Telepítsd az előkiadást

Mivel az SQLAlchemy 2.1 béta verzióban van, pip install sqlalchemy alapértelmezés szerint telepíti a legfrissebb stabil 2.0.x kiadást. Telepítsd kifejezetten az előzetes kiadást:

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

Ellenőrizze a telepített verziót:

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

Kapcsolati URL-ek

Az mssql-python dialektust mssql+mssqlpython használja URL sémaként. Az általános formátum a következő:

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

SQL-hitelesítés

SQL hitelesítéshez a felhasználónevet és jelszót a kapcsolati URL-ben foglaljuk be:

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-hitelesítés

Microsoft Entra hitelesítéshez használj üres felhasználónevet és a authentication lekérdezési paramétert:

from sqlalchemy import create_engine

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

Megjegyzés:

ActiveDirectoryDefault használ DefaultAzureCredential, amely több hitelesítésszolgáltatót próbál egymás után. Az első kapcsolat lassú lehet, mert az SDK végigjárja a láncot, amíg meg nem talál egy működő szolgáltatót. A termelésben, ha tudod, melyik hitelesítéstípust használja a környezeted, közvetlenül megadd (például ActiveDirectoryMSI menedzselt identitásnál), hogy elkerüld a láncos sétát. További információ: Microsoft Entra-hitelesítés.

URL-ek programozott építése

A manuális URL-kódolás elkerüléséhez használja a(z) sqlalchemy.engine.URL.create elemet:

from sqlalchemy.engine import URL

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

Definiáljuk az ORM modelleket

Használd az SQLAlchemy deklaratív leképezését olyan modellek definiálásához, amelyek a Microsoft SQL táblákhoz hasonlítottak.

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

Jótanács

A Microsoft SQL az oszlopok automatikus növelésére használjaIDENTITY. Az SQLAlchemy ezt automatikusan leképezi egész számú elsődleges kulcsoszlopokhoz. A fentebb bemutatott explicit Identity() használata nem kötelező, hacsak nem kell megadnia a kezdő- és növekményértékeket.

CRUD műveletek

Az alábbi példák bemutatják, hogyan lehet sorokat hozzáadni, lekérdezni, frissíteni és törölni az ORM ülés használatával. Minden példa újra felhasználja a new_id elemet, vagyis a ProductID értéket, amelyet egy sor beszúrásakor kap. A négy művelet együttes futtatásához lásd a teljes példát.

Munkamenet létrehozása

Hozz létre egy ülést a tranzakción belüli műveletek végrehajtásához:

from sqlalchemy.orm import Session

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

Olyan alkalmazásokhoz, amelyek sok munkamenetet hoznak létre, használd sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Sorok beszúrása

Új termék hozzáadása, a munkamenet elköteleződése, és a generált ProductID adatok rögzítése az alábbi példákhoz:

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

Megjegyzés:

A(z) SalesLT.Product esetében mind a Name, mind a ProductNumber egyedi megkötésekkel rendelkezik. Ha ezt a betétet többször futtatod, először változtasd meg ezeket az értékeket, vagy töröld az előző sort. A teljes példa törli a létrehozott sort, így ismételten futhat.

Lekérdezéssorok

Egyetlen sor lekérése elsődleges kulcs alapján, vagy szűrt lekérdezésekhez való felhasználás 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}")

Sorok frissítése

Módosítsd egy meglévő sor mezőjét, és véglegesítsd:

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

Sorok törlése

Távolíts el egy sort, és véglegesítsd:

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

Teljes példa

Az előző részek minden darabot külön mutattak be. Ez a rész egyesíti őket egy önálló szkriptté, amit lemásolhatsz, futtathatsz és újra futtathatsz.

Hozzon létre egy elnevezett crud.py fájlt, és adja hozzá a következő kódot. Cseréld ki a kapcsolati adatokat create_engine a sajátoddal (lásd Kapcsolati URL-ek):

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

Futtassa a szkriptet:

python crud.py

Hasonló kimenetet látsz, mint az alábbi:

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

A szkript törli az általa létrehozott sort, így amikor újra futtatja, nem ütközik a Name és ProductNumber egyediségi megkötéseibe. Minden egyes futtatás új sort szúr be, így a ProductID minden alkalommal növekszik.

Maglekérdezések

SQLAlchemy A Core alacsonyabb szintű SQL kifejezési API-t biztosít. Használhatod a Core-t ugyanazzal a motorral és tábladefinícióval, beleértve az ORM-térképes osztályokat is.

from sqlalchemy import text

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

Használjon táblázatszintű konstrukciókat típusbiztonsági SQL generáláshoz:

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

Megjegyzés:

Amikor olyan egyedileg leképezett oszlopokat választ ki, amelyeknek az adatbázisbeli neve eltér az attribútum nevétől (például a(z) Product.name a(z) Name oszlopra van leképezve), a Core sorai az adatbázisoszlop neve alapján vannak kulcsolva. Adja hozzá a(z) .label("name") elemet, hogy az értéket row.name-ként érje el row.Name helyett.

Kapcsolatmegosztás

Az SQLAlchemy alapértelmezés szerint egy kapcsolati poolt kezel. Hangold a pool beállításait a munkaterhelésedhez:

engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost/<database>",
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=3600,
)
Paraméter Leírás
pool_size A nyitva tartandó kapcsolatok száma (alapértelmezett: 5).
max_overflow A(z) pool_size feletti kapcsolatok engedélyezettek (alapértelmezett: 10).
pool_timeout Másodpercek a kapcsolatra várni, mielőtt hiba keletkezik (alapértelmezett: 30).
pool_recycle Másodpercekkel később újrahasznosítják a kapcsolatot (alapértelmezett: -1, kikapcsolva). Állítsd be ezt az értéket, ha az adatbázisod lezárja az üres kapcsolatokat.

Használat webes keretrendszerekkel

Az SQLAlchemy gyakran használják adatbázis-rétegként a Flask és a FastAPI számára. Az mssql-python dialektus bármely olyan keretrendszerrel működik, amely támogatja az SQLAlchemy-t.

A következő részletek bemutatják az ajánlott session-per request mintát minden keretrendszerhez. Ezek szemléltető részletek, amelyek a korábbi szakaszokban ismertetett engine és Product modellt feltételezik, nem teljes alkalmazások. Teljes, futtatható alkalmazásokért lásd a FastAPI integrációs és Flask integrációs cikkeket.

FastAPI példa

Használj generátoros függőséget, hogy minden kéréshez egy munkamenetet biztosíts:

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

Flaska példa

Használj kontextuskezelőt, hogy a munkamenetet a kéréshez igazítsd:

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

Alembik vándorlások

Alembic kezeli a SQLAlchemy-projektek sémamigrációit, és az mssql-python dialektussal működik. Az Alembic automatikus generáló funkciója összehasonlítja a modelljeidet az élő adatbázissal, így néhány plusz lépés megakadályozza, hogy olyan táblázatokban javasoljon változtatásokat, amelyeket nem kezelsz.

Alembic létrehozása

Telepítsd az Alembic-et és inicializáld a migrációs könyvtárat:

pip install alembic
alembic init migrations

A ben alembic.ini, állítsd be a kapcsolati URL-t:

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

Irányítsd az Alembicet a modelljeidre

Az Autogeneration-nak szüksége van a modelljeid metaadataira. Tedd az Alembic által kezelt modelleket egy importálható modulba, például models.py. Mivel az autogenerálás azt javasolja, hogy bármely oszlopot elhagyjon, amelyet egy modell kihagy, definiáljunk egy olyan modellt, amely teljes mértékben birtokolja a tábláját, ahelyett, hogy újra használnád a cikkben korábban említett egyszerűsített Product modellt:

# 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

Alapértelmezés szerint az automatikus generálás az adatbázis minden olyan tábláját, amely nem szerepel a(z) target_metadata elemben, eltávolítottnak tekinti, és drop_table utasítást hoz létre hozzá. Egy meglévő adatbázis, például az AdventureWorksLT esetében ez a művelet több tucat táblát is törölhet. Adj egy include_name szűrőt, hogy az Alembic csak azokat a táblázatokat kezelje, amiket a modelljeid definiálnak, és mindig nézze át a generált szkriptet, mielőtt alkalmaznád.

A(z) migrations/env.py elemben cseréld le a(z) target_metadata = None elemet a következő kódra. Importálja a modelleket, és az automatikus generálást az általuk meghatározott sémákra és táblákra korlátozza:

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

Add át a(z) include_name és include_schemas=True elemeket a(z) context.configure számára mind a(z) run_migrations_offline, mind a(z) run_migrations_online esetében. A include_schemas=True beállítás lehetővé teszi, hogy az Alembic nem alapértelmezett sémákban is lásson táblázatokat, például SalesLT:

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

Generálj és alkalmazz migrációt

Generálj migrációt a modelljeidből:

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

Az Alembic felismeri az új táblát, és migrációs szkriptet ír:

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

A generált upgrade() létrehozza a táblázatot, majd downgrade() eldobja:

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

Nézd át a szkriptet, majd alkalmazd az összes függőben lévő migrációt:

alembic upgrade head

Különbségek a pyodbc dialektustól

Ha a(z) mssql+pyodbc rendszerről migrálsz, az mssql-python dialektus hasonló lesz, mivel mindkét illesztőprogram ugyanarra az ODBC keretrendszerre épül. Főbb különbségek:

Téma mssql+pyodbc mssql+mssqlpython
ODBC-illesztőprogram telepítése Külön ODBC illevezetőt igényel (például ODBC Driver 18 Microsoft SQL-hez). Az illesztőprogram mellékelve van. Nem kell külön ODBC driver.
Kapcsolat URL mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server mssql+mssqlpython://user:pass@host/db
fast_executemany Támogatva a következőn keresztül: create_engine(..., fast_executemany=True). Nem alkalmazható. A meghajtó belső szakaszteljesítményt kezeli.
Availability Stabil, az SQLAlchemy-be van beépítve az 1.x óta. Előzetes kiadás (SQLAlchemy 2.1.0b2+).

Ismert korlátozások

Az mssql-python dialektusa SQLAlchemy számára előzetes kiadásban van. Mielőtt gyártásban használnánk, értsd meg ezeket a következményeket:

  • API változások: A metódusaláírások, kivételtípusok és viselkedés változhatnak a végleges stabil kiadás előtt. Mindig rögzítsd az SQLAlchemy verziódat egy adott előzetes verzióhoz (például), sqlalchemy==2.1.0b2és alaposan teszteld a frissítéseket.

  • Korlátozott tesztelés: A dialektusban kevesebb közösségi tesztelés van, mint a stabil mssql+pyodbc dialektus. Előfordulhat, hogy szélső esetek vagy hiányzó funkciók is előfordulhatnak.

  • Funkcióhiányok: Néhány fejlett ORM vagy Core funkció nem feltétlenül működik. Nézd meg az SQLAlchemy MSSQL dialektus dokumentációt , és teszteld a felhasználási eseteidet, mielőtt elköteleződnél egy projekthez.

  • Nincs támogatási garancia: a Microsoft és az SQLAlchemy a lehető legjobb támogatást nyújtja, de a problémák nem feltétlenül oldódnak meg a stabil kiadás előtt.

Mikor érdemes használni az előzetes kiadást:

  • Fejlesztési és tesztelési környezetek
  • Koncepció-bizonyítási projektek
  • Áttérés erről: mssql+pyodbc, ha el szeretné kerülni a külső ODBC-illesztőprogram-függőséget
  • Olyan projektek, ahol reagálhatsz API-változásokra és regressziós tesztelést végezhetsz

Mikor NE használd az előzetes kiadást:

  • Szigorú stabilitási követelményeket igénylő gyártási rendszerek
  • Többéves örökségi alkalmazások, ahol a függőségi frissítések ritkák
  • Kritikus üzleti munkaterhelések, amíg az SQLAlchemy 2.1 stabil GA-t nem éri el

A legfrissebb kiadás előtti dialektus állapotért és ismert problémákért nézd meg az mssql-python GitHub repozitóriumot.

Hibaelhárítás

Nincs „sqlalchemy.dialects.mssql.mssqlpython” néven modul.

Ez a hiba azt jelenti, hogy a telepített SQLAlchemy verzió nem tartalmazza az mssql-python dialektust. Ellenőrizd, hogy 2.1.0b2 vagy újabb verziód van:

pip install "sqlalchemy>=2.1.0b2"

Csatlakozási hibák

Ha create_engine sikerül, de a lekérdezések kudarcosak, ellenőrizd, hogy a kapcsolati paramétereid közvetlenül az mssql-pythonnal működnek-e:

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

Ha a közvetlen kapcsolat működik, de a SQLAlchemy nem, ellenőrizd, hogy nincs-e URL-kódolási probléma a jelszóban vagy a szervernévben szereplő speciális karakterekkel.