Použijte mssql-python s SQLAlchemy

SQLAlchemy je nejrozšířenější Python ORM a databázový toolkit. Počínaje verzí SQLAlchemy 2.1.0b2 je k dispozici vestavěný dialekt pro ovladač mssql-python, který umožňuje používat SQLAlchemy ORM a Core pro Microsoft SQL a Azure SQL Database.

Important

Dialekt mssql-python byl přidán v SQLAlchemy 2.1.0b2 (vydáno 16. dubna 2026). SQLAlchemy 2.1 je momentálně sérií předběžných verzí a nedoporučuje se pro produkční nasazení. Před upgradem ze SQLAlchemy 2.0 si povědomě:

• API se mohou změnit před finálním stabilním vydáním (2.1 GA)
• Před nasazením důkladně otestujte svou pracovní zátěž
• Používejte stabilní SQLAlchemy 2.0.x pro produkční systémy, dokud 2.1 nedosáhne GA
• Připněte svou závislost na konkrétní verzi (například sqlalchemy==2.1.0b2) místo používání rozsahů verzí

Podrobnosti o tom, kdy používat předběžné verze, viz sekce Známá omezení .

Předpoklady

• Python 3.10 nebo novější. SQLAlchemy 2.1 zrušil podporu pro Python 3.9 a starší modely.
• Balíčky mssql-python and sqlalchemy (2.1.0b2 nebo později).

Příklady v tomto článku využívají vzorovou databázi AdventureWorksLT . Pokud nemáte AdventureWorksLT nainstalovaný, podívejte se na ukázkové databáze AdventureWorks.

Nainstalujte předběžné vydání

Protože SQLAlchemy 2.1 je v beta verzi, instaluje pip install sqlalchemy nejnovější stabilní verzi 2.0.x ve výchozím nastavení. Předběžné vydání nainstalujte explicitně:

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

Ověřte nainstalovanou verzi:

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

Adresy URL připojení

Dialekt mssql-python používá mssql+mssqlpython jako schéma URL. Obecný formát je:

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

Ověřování SQL

Pro ověření SQL uveďte uživatelské jméno a heslo do URL připojení:

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

Ověřovací systém Microsoft Entra

Pro autentizaci Microsoft Entra použijte prázdné uživatelské jméno a authentication parametr dotazu:

from sqlalchemy import create_engine

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

Note

ActiveDirectoryDefault používá DefaultAzureCredential, která zkouší více poskytovatelů přihlašovacích údajů postupně. První spojení může být pomalé, protože SDK prochází řetězec, dokud nenajde funkčního poskytovatele. V produkci, pokud víte, jaký typ přihlašovacích údajů vaše prostředí používá, zadejte ho přímo (například ActiveDirectoryMSI pro spravovanou identitu), abyste se vyhnuli tzv. chain walk. Další informace naleznete v tématu ověřování Microsoft Entra.

Vytváření URL programově

Použijte sqlalchemy.engine.URL.create pro vyhnutí se ručnímu kódování URL:

from sqlalchemy.engine import URL

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

Definujte modely ORM

Použijte deklarativní mapování SQLAlchemy k definování modelů, které se mapují na tabulky Microsoft SQL.

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 používá IDENTITY automatické inkrementování sloupců. SQLAlchemy to automaticky mapuje pro sloupce celočíselného primárního klíče. Explicitní Identity() výše uvedené je volitelné, pokud nepotřebujete ovládat počáteční a inkrementní hodnoty.

operace CRUD

Následující příklady ukazují, jak vkládat, dotazovat, aktualizovat a mazat řádky pomocí relace ORM. Každý příklad znovu používá new_id, tedy ProductID, které se vrátí při vložení řádku. Pro spuštění všech čtyř operací současně viz kompletní příklad.

Vytvořte relaci

Vytvořte relaci pro provádění operací v rámci transakce:

from sqlalchemy.orm import Session

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

Pro aplikace, které vytvářejí velké množství relací, použijte sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Vložte řádky

Přidejte nový produkt, potvrďte relaci a zaznamenejte vygenerované ProductID pro následující příklady:

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

V SalesLT.Product mají jak Name, tak ProductNumber jedinečná omezení. Pokud tuto vložku spustíte vícekrát, změňte tyto hodnoty nebo nejprve smažte předchozí řádek. Kompletní příklad vymaže řádek, který vytvoří, takže může běžet opakovaně.

Řádky dotazu

Získejte jeden řádek pomocí primárního klíče, nebo použijte select() pro filtrované dotazy:

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

Aktualizace řádků

Upravte pole ve stávajícím řádku a potvrďte změny:

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

Odstranit řádky

Odstraňte řádek a potvrďte změny:

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

Kompletní příklad

Předchozí sekce ukazovaly každý kus zvlášť. Tato sekce je spojuje do jednoho samostatného skriptu, který můžete kopírovat, spouštět a znovu spouštět.

Vytvořte soubor s názvem crud.py a přidejte následující kód. Nahraďte podrobnosti create_engine spojení svými vlastními (viz URL připojení):

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

Spusťte skript:

python crud.py

Vidíte výstup podobný následujícímu:

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

Skript odstraní řádek, který vytvoří, takže při opětovném spuštění nenarazí na unikátní omezení u Name a ProductNumber Každý běh vkládá nový řádek, takže se počet ProductID zvyšuje pokaždé.

Základní dotazy

SQLAlchemy Core poskytuje nízkoúrovňové SQL expression API. Core můžete používat se stejným enginem a definicemi tabulek, včetně tříd mapovaných ORM.

from sqlalchemy import text

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

Používejte tabulkové konstrukce pro typově bezpečné generování SQL:

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

Když vyberete jednotlivé mapované sloupce, jejichž název databáze se liší od názvu atributu (například mapuje Product.name na Name sloupec), základní řádky jsou zadány názvem sloupce databáze. Přidejte .label("name"), abyste přistupovali k hodnotě jako row.name místo row.Name.

Sdílení připojení

SQLAlchemy ve výchozím nastavení spravuje pool připojení. Nastavte nastavení poolu podle své pracovní zátěže:

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

Parameter	Popis
pool_size	Počet připojení, která je třeba nechat otevřená (výchozí: 5).
max_overflow	Povolena připojení za hranicí pool_size (výchozí: 10).
pool_timeout	Několik sekund čekání na připojení, než se objeví chyba (výchozí: 30).
pool_recycle	Sekundy po kterých je spojení recyklováno (výchozí: -1, deaktivováno). Nastavte tuto hodnotu, pokud vaše databáze uzavírá nečinné připojení.

Použití s webovými frameworky

SQLAlchemy se běžně používá jako databázová vrstva pro Flask a FastAPI. Dialekt mssql-python funguje s jakýmkoli frameworkem, který podporuje SQLAlchemy.

Následující úryvky ukazují doporučený vzor relace na požadavek pro každý rámec. Jsou to ilustrativní fragmenty, které vycházejí z modelu engine a Product popsaného v předchozích oddílech, nikoli kompletní aplikace. Pro kompletní, spuštěné aplikace viz články o integraci FastAPI a Flask .

Příklad FastAPI

Použijte závislost založenou na generátoru k poskytování relace pro každý požadavek:

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

Příklad ve Flasku

Pomocí správce kontextu omezte relaci na dobu zpracování požadavku:

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

Alembické migrace

Alembic řeší migrace schémat pro projekty SQLAlchemy a pracuje s dialektem mssql-python. Funkce automatického generování v Alembicu porovnává vaše modely s živou databází, takže pár dalších kroků zabraňuje tomu, aby navrhoval změny v tabulkách, které nespravujete.

Založte Alembic

Nainstalujte Alembic a inicializujte adresář migrací:

pip install alembic
alembic init migrations

V alembic.ini, nastavte URL spojení:

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

Nasměrujte Alembic na své modely

Autogenerate potřebuje metadata vašich modelů. Umístěte modely, které Alembic spravuje, do importovatelného modulu, například models.py. Protože funkce autogenerate navrhuje odstranit jakýkoli sloupec, který model nezahrnuje, definujte model, který plně pokrývá svou tabulku, namísto opětovného použití zjednodušeného modelu Product z dřívější části tohoto článku:

# 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

Ve výchozím nastavení autogenerate považuje každou tabulku v databázi, která není v target_metadata, za odstraněnou a vygeneruje pro ni drop_table. V existující databázi, například AdventureWorksLT, může tato akce odstranit desítky tabulek. Přidejte filtr, include_name aby Alembic spravoval pouze tabulky, které vaše modely definují, a vždy si před aplikací zkontrolujte vygenerovaný skript.

V , migrations/env.pynahraďte target_metadata = None následujícím kódem. Importuje vaše modely a omezuje automatické generování na schémata a tabulky, které definují:

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

Předejte include_name a include_schemas=True do context.configure v run_migrations_offline i run_migrations_online. Toto include_schemas=True nastavení umožňuje Alembic vidět tabulky v nevýchozích schématech, jako jsou SalesLT:

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

Vygenerujte a aplikujte migraci

Generujte migraci ze svých modelů:

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

Alembic detekuje novou tabulku a napíše skript pro migraci:

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

Vygenerovaný upgrade() vytvoří tabulku a downgrade() ji odstraní:

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

Zkontrolujte skript a pak spusťte všechny čekající migrace:

alembic upgrade head

Rozdíly oproti pyodbcskému dialektu

Pokud migrujete z mssql+pyodbc, je dialekt mssql-python podobný, protože oba ovladače jsou založeny na stejném rámci ODBC. Hlavní rozdíly:

Téma	mssql+pyodbc	mssql+mssqlpython
Instalace ovladačů ODBC	Vyžaduje samostatný ODBC ovladač (například ODBC Driver 18 pro Microsoft SQL).	Ovladač je součástí balíčku. Není potřeba žádný samostatný ODBC ovladač.
Adresa URL připojení	mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server	mssql+mssqlpython://user:pass@host/db
fast_executemany	Podporováno prostřednictvím create_engine(..., fast_executemany=True).	Nelze použít. Ovladač interně řeší výkon dávkového zpracování.
Availability	Stabilní, zahrnuté v SQLAlchemy od verze 1.x.	Předběžná verze (SQLAlchemy 2.1.0b2+).

Známá omezení

Dialekt mssql-python pro SQLAlchemy je ve fázi předběžné verze. Před použitím ve výrobě si povědomte tyto důsledky:

• Změny API: Podpisy metod, typy výjimek a chování se mohou změnit před finálním stabilním vydáním. Vždy připněte svou verzi SQLAlchemy ke specifické předběžné verzi (například) sqlalchemy==2.1.0b2a důkladně testujte upgrady.

• Omezené testování: Dialekt má méně komunitního testování než stabilní dialekt mssql+pyodbc . Můžete narazit na krajní případy nebo chybějící funkce.

• Nedostatky ve funkcích: Některé pokročilé funkce ORM nebo Core nemusí fungovat. Podívejte se na dokumentaci k dialektu SQLAlchemy MSSQL a otestujte své případy použití, než se zavážete k projektu.

• Žádná záruka podpory: Microsoft a SQLAlchemy poskytují podporu na základě nejlepšího úsilí, ale problémy nemusí být vyřešeny před stabilním vydáním.

Kdy použít předběžné vydání:

• Vývojová a testovací prostředí
• Projekty důkazu konceptu
• Migrace z mssql+pyodbc, pokud se chcete vyhnout závislosti na externím ovladači ODBC
• Projekty, kde můžete reagovat na změny API a provádět regresní testování

Kdy NEPOUŽÍVAT předběžné vydání:

• Výrobní systémy s přísnými požadavky na stabilitu
• Víceroční legacy aplikace, kde jsou aktualizace závislostí vzácné
• Kritické obchodní zátěže, dokud SQLAlchemy 2.1 nedosáhne stabilní GA

Pro nejnovější stav předvydání dialektů a známé problémy se podívejte do repozitáře mssql-python GitHub.

Řešení problémů

"Žádný modul s názvem 'sqlalchemy.dialects.mssql.mssqlpython'"

Tato chyba znamená, že vaše nainstalovaná verze SQLAlchemy neobsahuje dialekt mssql-python. Ověřte, že máte verzi 2.1.0b2 nebo novější:

pip install "sqlalchemy>=2.1.0b2"

Selhání připojení

Pokud create_engine uspěje, ale dotazy selžou, ověřte, že vaše parametry připojení fungují přímo s 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()

Pokud přímé spojení funguje, ale SQLAlchemy ne, zkontrolujte problémy s kódováním URL speciálními znaky v hesle nebo jménu serveru.

Související obsah

• MSSQL-Python Connection Strings
• MSSQL-Python Connection Management
• Integrace s FastAPI
• Integrace s baňkou
• Dokumentace dialektu SQLAlchemy MSSQL