Použijte mssql-python s SQLAlchemy

SQLAlchemy je nejrozšířenější Python ORM a databázový toolkit. SQLAlchemy 2.1 obsahuje vestavěný dialekt pro ovladač mssql-python, který můžete použít pro práci s moduly ORM a Core v SQLAlchemy pro Microsoft SQL a Azure SQL Database.

Předpoklady

Python 3.11 nebo novější. SQLAlchemy 2.1 zrušil podporu pro Python 3.10 a starší formáty.
Balíčky mssql-python a sqlalchemy (verze 2.1 nebo novější).

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 SQLAlchemy a mssql-python

Použijte volitelnou závislost SQLAlchemymssql-python k instalaci obou balíčků:

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

SQLAlchemy definuje extra mssql-python pomocí mssql-python>=1.9.0. Podporovaná je také existující forma samostatného balíčku:

pip install "sqlalchemy>=2.1" mssql-python

Ověřte nainstalovanou verzi:

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

Upozornění

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:

Topic	`mssql+pyodbc`	`mssql+mssqlpython`
Instalace ovladačů ODBC	Vyžaduje samostatný ODBC ovladač (například ODBC Driver 18 pro SQL Server).	Nainstaluje se automaticky jako závislost balíčku. Není potřeba žádná samostatná instalace ovladačů ODBC.
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.	Stabilní, zahrnuté v SQLAlchemy 2.1 nebo novějším.

Úvahy o upgradu

SQLAlchemy 2.1 zahrnuje změny chování, které mohou ovlivnit aktualizaci aplikací oproti dřívějším verzím:

SQLAlchemy 2.1 vyžaduje Python 3.11 nebo starší.
Upgradujte aplikace SQLAlchemy 1.x na SQLAlchemy 2.0, než přejděte na 2.1.
Otestujte stávající aplikace SQLAlchemy 2.0 proti změnám chování popsaným v článku Co je nového v SQLAlchemy 2.1?

Příklady v tomto článku využívají synchronní API SQLAlchemy. Aplikace, které používají podporu SQLAlchemy asyncio, musí nainstalovat doplněk asyncio, protože SQLAlchemy 2.1 již standardně neinstaluje greenlet.

Troubleshooting

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

Tato chyba znamená, že vaše nainstalovaná verze SQLAlchemy dialekt mssql-python neobsahuje. Nainstalujte SQLAlchemy 2.1 nebo novější včetně závislosti mssql-python:

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

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.

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-09-25

Použijte mssql-python s SQLAlchemy

Předpoklady

Nainstalujte SQLAlchemy a mssql-python

Adresy URL připojení

Ověřování SQL

Ověřovací systém Microsoft Entra

Vytváření URL programově

Definujte modely ORM

operace CRUD

Vytvořte relaci

Vložte řádky

Řádky dotazu

Aktualizace řádků

Odstranit řádky

Kompletní příklad

Základní dotazy

Sdílení připojení

Použití s webovými frameworky

Příklad FastAPI

Příklad ve Flasku

Alembické migrace

Založte Alembic

Nasměrujte Alembic na své modely

Vygenerujte a aplikujte migraci

Rozdíly oproti pyodbcskému dialektu

Úvahy o upgradu

Troubleshooting

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

Selhání připojení

Související obsah

Váš názor

Další materiály