Użyj mssql-python z SQLAlchemy

SQLAlchemy to najczęściej używany zestaw narzędzi Python ORM i baz danych. Zaczynając od SQLAlchemy 2.1.0b2, wbudowany dialekt sterownika mssql-python pozwala używać SQLAlchemy ORM i Core z Microsoft SQL i Azure SQL Database.

Important

Dialekt mssql-python został dodany w SQLAlchemy 2.1.0b2 (wydanym 16 kwietnia 2026). SQLAlchemy 2.1 jest obecnie serią wydań przedpremierowych i nie jest zalecana do użytku produkcyjnego. Przed przejściem na SQLAlchemy 2.0 warto zrozumieć:

• API mogą się zmieniać przed ostateczną stabilną wersją (2.1 GA)
• Dokładnie przetestuj swoje obciążenie przed wdrożeniem
• Używaj stabilnej wersji SQLAlchemy 2.0.x w systemach produkcyjnych do czasu, aż wersja 2.1 osiągnie ogólną dostępność (GA)
• Przypnij swoją zależność do konkretnej wersji (na przykład sqlalchemy==2.1.0b2) zamiast używać zakresów wersji

Zobacz sekcję Znane Ograniczenia , aby poznać szczegóły dotyczące stosowania wersji przedpremierowych.

Wymagania wstępne

• Python 3.10 lub nowszy. SQLAlchemy 2.1 zrezygnowało z obsługi Python 3.9 i wcześniejszych wersji.
• Pakiety mssql-python and sqlalchemy (2.1.0b2 lub nowsze).

Przykłady w tym artykule wykorzystują przykładową bazę danych AdventureWorksLT . Jeśli nie masz zainstalowanego AdventureWorksLT, zobacz przykładowe bazy danych AdventureWorks.

Zainstaluj wersję przedpremierową

Ponieważ SQLAlchemy 2.1 jest w fazie beta, domyślnie instaluje pip install sqlalchemy najnowszą stabilną wersję 2.0.x. Zainstaluj jawnie wersję przedpremierową w ten sposób:

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

Sprawdź zainstalowaną wersję:

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

Adresy URL połączeń

Dialekt mssql-python używa mssql+mssqlpython jako schematu URL. Ogólny format to:

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

Uwierzytelnianie SQL

Do uwierzytelniania SQL należy podać nazwę użytkownika i hasło w adresie URL połączenia:

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

Uwierzytelnianie usługi Microsoft Entra

Do uwierzytelniania Microsoft Entra użyj pustej nazwy użytkownika oraz parametru authentication zapytania:

from sqlalchemy import create_engine

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

Note

ActiveDirectoryDefault używa DefaultAzureCredential, który testuje kolejno wielu dostawców poświadczeń. Pierwsze połączenie może być wolne, ponieważ SDK przechodzi przez łańcuch, aż znajdzie dostawcę, który działa. W środowisku produkcyjnym, jeśli wiesz, jakiego typu poświadczeń używa środowisko, wskaż go bezpośrednio (na przykład ActiveDirectoryMSI w przypadku tożsamości zarządzanej), aby uniknąć przechodzenia przez łańcuch. Aby uzyskać więcej informacji, zobacz Microsoft Entra authentication (Uwierzytelnianie w usłudze Microsoft Entra).

Buduj adresy URL programatycznie

Użycie metody sqlalchemy.engine.URL.create , aby uniknąć ręcznego kodowania 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)

Zdefiniuj modele ORM

Wykorzystaj deklaratywne mapowanie SQLAlchemy, aby zdefiniować modele odwzorowane na tabelach 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()
    )

Wskazówka

Microsoft SQL używa IDENTITY dla kolumn autoinkrementowanych. SQLAlchemy automatycznie mapuje to dla kolumn klucza głównego typu całkowitego. Pokazane powyżej jawnie określone Identity() jest opcjonalne, chyba że musisz określić wartości początkową i przyrostową.

Operacje CRUD

Poniższe przykłady pokazują, jak wstawiać, zapytywać, aktualizować i usuwać wiersze za pomocą sesji ORM. Każdy przykład ponownie wykorzystuje new_id, a zwraca ProductID się, gdy wstawiasz wiersz. Aby wykonać wszystkie cztery operacje razem, zobacz pełny przykład.

Tworzenie sesji

Utwórz sesję do wykonania operacji w ramach transakcji:

from sqlalchemy.orm import Session

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

Dla aplikacji tworzących wiele sesji użyj sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Wstawianie wierszy

Dodaj nowy produkt, zatwierdź sesję i przechwyć wygenerowane ProductID dla następujących przykładów:

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

W SalesLT.Product zarówno Name, jak i ProductNumber mają unikalne ograniczenia. Jeśli uruchomisz tę wstawkę więcej niż raz, zmień te wartości lub usuń najpierw wcześniejszy wiersz. Pełny przykład usuwa utworzony przez niego wiersz, dzięki czemu może działać wielokrotnie.

Wiersze zapytań

Pobierz pojedynczy wiersz za pomocą klucza głównego lub użyj select() do zapytań filtrowanych:

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

Zaktualizuj wiersze

Zmodyfikuj pole w istniejącym wierszu i zatwierdź:

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

Usuwanie wierszy

Usuń wiersz i zatwierdź:

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

Kompletny przykład

Poprzednie sekcje pokazywały każdy utwór osobno. Ta sekcja łączy je w jeden samodzielny skrypt, który możesz skopiować, uruchomić i uruchomić ponownie.

Utwórz plik o nazwie crud.py i dodaj następujący kod. Zamień szczegóły create_engine połączenia na swoje (patrz Adresy URL połączeń):

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

Uruchom skrypt:

python crud.py

Widzisz wyniki zbliżone do następujących:

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

Skrypt usuwa wiersz, który tworzy, więc przy Name ponownym uruchomieniu ProductNumber nie narusza unikalnych ograniczeń. Każde uruchomienie wstawia nowy wiersz, więc ProductID zwiększa się za każdym razem.

Zapytania podstawowe

SQLAlchemy Core oferuje niskopoziomowe API do wyrażania SQL. Możesz używać Core z tym samym silnikiem i definicjami tabel, w tym klasami mapowanymi przez ORM.

from sqlalchemy import text

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

Używaj konstrukcji na poziomie tabeli do generowania SQL bezpiecznego dla typów:

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

Po wybraniu pojedynczych mapowanych kolumn, których nazwa w bazie danych różni się od nazwy atrybutu (na przykład elementowi Product.name odpowiada kolumna Name), wiersze Core są identyfikowane według nazwy kolumny bazy danych. Dodaj .label("name"), aby uzyskać dostęp do wartości jako row.name zamiast row.Name.

Buforowanie połączeń

SQLAlchemy domyślnie zarządza pulą połączeń. Dostosuj ustawienia puli do swojego obciążenia:

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

Parameter	Opis
pool_size	Liczba połączeń do utrzymania otwartej (domyślnie: 5).
max_overflow	Połączenia dozwolone powyżej pool_size (domyślnie: 10).
pool_timeout	Sekundy na oczekiwanie na połączenie przed pojawieniem się błędu (domyślnie: 30).
pool_recycle	Liczba sekund, po których połączenie jest resetowane (domyślnie: -1, wyłączone). Ustaw tę wartość, jeśli baza danych zamyka bezczynne połączenia.

Zastosowanie z frameworkami webowymi

SQLAlchemy jest powszechnie używany jako warstwa bazy danych dla Flask i FastAPI. Dialekt mssql-python działa z każdym frameworkiem obsługującym SQLAlchemy.

Poniższe fragmenty pokazują zalecany wzorzec sesji na żądanie dla każdego frameworka. To przykładowe fragmenty, które zakładają model engineProduct z poprzednich sekcji, a nie są kompletnymi aplikacjami. Aby poznać kompletne, możliwe do uruchomienia aplikacje, zobacz artykuły o integracji FastAPI i Flask .

Przykład FastAPI

Użyj zależności generatora, aby zapewnić sesję na każde żądanie:

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

Przykład Flaska

Użyj menedżera kontekstu, aby ograniczyć zakres sesji do żądania:

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

Migracje alembików

Alembic obsługuje migracje schematów dla projektów SQLAlchemy i współpracuje z dialektem mssql-python. Funkcja automatycznego generowania w Alembic porównuje twoje modele z bazą danych na żywo, więc kilka dodatkowych kroków zapobiega proponowaniu zmian w tabelach, których nie obsługujesz.

Załóż Alembic

Zainstaluj Alembic i zainicjalizuj katalog migracji:

pip install alembic
alembic init migrations

W alembic.ini, ustaw adres URL połączenia:

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

Skieruj Alembic na swoje modele

Autogenerowanie wymaga metadanych twoich modeli. Umieść modele, którymi zarządza Alembic, w module importowalnym, takim jak models.py. Ponieważ autogenerate proponuje usunięcie każdej kolumny, którą model pomija, zdefiniuj model, który w pełni kontroluje swoją tabelę, zamiast ponownie używać uproszczonego Product modelu z wcześniejszego artykułu:

# 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

Domyślnie funkcja autogenerate traktuje każdą tabelę w bazie danych, której nie ma w target_metadata, jako usuniętą i generuje dla niej drop_table. W porównaniu z istniejącą bazą danych taką jak AdventureWorksLT, takie działanie może zlikwidować dziesiątki tabel. Dodaj filtr, include_name aby Alembic zarządzał tylko tabelami zdefiniowanymi przez twoje modele i zawsze przeglądaj wygenerowany skrypt przed jego zastosowaniem.

W migrations/env.py zastąp target_metadata = None następującym kodem. Importuje twoje modele i ogranicza automatyczne generowanie do schematów i tabel, które definiują:

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

Przekaż include_name i include_schemas=True do context.configure zarówno w run_migrations_offline, jak i w run_migrations_online. Ustawienie include_schemas=True umożliwia narzędziu Alembic wykrywanie tabel w schematach innych niż domyślny, takich jak SalesLT:

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

Wygeneruj i zastosuj migrację

Wygeneruj migrację na podstawie modeli:

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

Alembic wykrywa nową tabelę i pisze skrypt migracji:

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

Wygenerowany upgrade() tworzy tabelę i downgrade() ją porzuca:

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

Przejrzyj skrypt, a następnie zastosuj wszystkie oczekujące migracje:

alembic upgrade head

Różnice w stosunku do dialektu pyodbc

Jeśli migrujesz z mssql+pyodbc, dialekt mssql-python jest podobny, ponieważ oba sterowniki opierają się na tym samym frameworku ODBC. Kluczowe różnice:

Temat	mssql+pyodbc	mssql+mssqlpython
Instalacja sterowników ODBC	Wymaga osobnego sterownika ODBC (na przykład ODBC Driver 18 dla Microsoft SQL).	Sterownik jest dołączony w pakiet. Nie potrzebny jest osobny sterownik ODBC.
Adres URL połączenia	mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server	mssql+mssqlpython://user:pass@host/db
fast_executemany	Wspierane przez create_engine(..., fast_executemany=True).	Nie dotyczy. Sterownik obsługuje wydajność wsadową wewnętrznie.
Dostępność	Stabilne, włączone do SQLAlchemy od wersji 1.x.	Przedpremiera (SQLAlchemy 2.1.0b2+).

Znane ograniczenia

Dialekt mssql-python dla SQLAlchemy jest dostępny w wersji prerelease. Przed użyciem w produkcji należy zrozum następujące konsekwencje:

• Zmiany API: Sygnatury metod, typy wyjątków i zachowanie mogą się zmieniać przed ostatecznym wydaniem stabilnym. Zawsze przypinaj swoją wersję SQLAlchemy do konkretnej wersji pre-release (na przykład) sqlalchemy==2.1.0b2i dokładnie testuj aktualizacje.

• Ograniczone testy: Dialekt ma mniej testów społecznych niż stabilny dialekt mssql+pyodbc . Możesz napotkać nietypowe przypadki lub brak niektórych funkcji.

• Braki funkcjonalne: Niektóre zaawansowane funkcje ORM lub Core mogą nie działać. Zapoznaj się z dokumentacją SQLAlchemy dotyczącą dialektu MSSQL i przetestuj swoje przypadki użycia przed podjęciem projektu.

• Brak gwarancji wsparcia: Microsoft i SQLAlchemy oferują wsparcie na poziomie najlepszego wysiłku, ale problemy mogą nie zostać rozwiązane przed stabilną wersją.

Kiedy używać wersji przedpremierowej:

• Środowiska programistyczne i testowe
• Projekty potwierdzające koncepcję
• Migracja z mssql+pyodbc, jeśli chcesz uniknąć zależności od zewnętrznego sterownika ODBC
• Projekty, w których możesz reagować na zmiany API i przeprowadzać testy regresyjne

Kiedy NIE używać wersji przedpremierowej:

• Systemy produkcyjne z rygorystycznymi wymaganiami dotyczącymi stabilności
• Wieloroczne aplikacje starsze, gdzie aktualizacje zależności są rzadkie
• Krytyczne obciążenia biznesowe do momentu, gdy SQLAlchemy 2.1 osiągnie stabilne GA

Aby poznać najnowszy status dialektu przed wydaniem i znane problemy, sprawdź repozytorium mssql-python GitHub.

Troubleshooting

"Brak modułu o nazwie 'sqlalchemy.dialects.mssql.mssqlpython'"

Ten błąd oznacza, że zainstalowana wersja SQLAlchemy nie zawiera dialektu mssql-python. Sprawdź, czy masz wersję 2.1.0b2 lub nowszą:

pip install "sqlalchemy>=2.1.0b2"

Awarie połączenia

Jeśli create_engine zadziała, ale wykonywanie zapytań się nie powiedzie, sprawdź, czy parametry połączenia działają bezpośrednio w 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()

Jeśli bezpośrednie połączenie działa, a SQLAlchemy nie, sprawdź problemy z kodowaniem URL w specjalnych znakach w hasle lub nazwie serwera.

Treści powiązane

• Łańcuchy połączeń mssql-python
• MSSQL-Python Connection Management
• Integracja z FastAPI
• Integracja z Flask
• Dokumentacja dialektu SQLAlchemy MSSQL