Użyj mssql-python z SQLAlchemy

SQLAlchemy to najczęściej używany zestaw narzędzi Python ORM i baz danych. SQLAlchemy 2.1 zawiera wbudowany dialekt dla sterownikamssql-python, który można wykorzystać do pracy z SQLAlchemy ORM oraz Core z Microsoft SQL i Azure SQL Database.

Wymagania wstępne

• Python 3.11 lub nowszy. SQLAlchemy 2.1 zrezygnowało z obsługi Python 3.10 i wcześniejszych wersji.
• Pakiety mssql-python i sqlalchemy (2.1 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 SQLAlchemy i mssql-python

Użyj opcjonalnej zależności SQLAlchemymssql-python, aby zainstalować oba pakiety:

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

SQLAlchemy definiuje mssql-python extra za pomocą mssql-python>=1.9.0. Obsługiwana jest także istniejąca forma pakietu oddzielnego:

pip install "sqlalchemy>=2.1" mssql-python

Sprawdź zainstalowaną wersję:

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

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

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

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

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

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

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:

Zagadnienia dotyczące uaktualniania

SQLAlchemy 2.1 zawiera zmiany zachowań, które mogą wpłynąć na aktualizacje aplikacji z wcześniejszych wersji:

• SQLAlchemy 2.1 wymaga Python 3.11 lub nowszy.
• Zaktualizuj aplikacje SQLAlchemy 1.x do SQLAlchemy 2.0, zanim przejdę na 2.1.
• Testuj istniejące aplikacje SQLAlchemy 2.0 z uwzględnieniem zmian zachowań opisanych w artykule Co jest nowe w SQLAlchemy 2.1?

Przykłady w tym artykule korzystają z synchronicznego API SQLAlchemy. Aplikacje korzystające z obsługi asyncio w SQLAlchemy muszą zainstalować pakiet extra asyncio, ponieważ SQLAlchemy 2.1 nie instaluje już domyślnie pakietu greenlet.

Troubleshooting

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

Ten błąd oznacza, że zainstalowana wersja SQLAlchemy nie zawiera dialektu mssql-python . Zainstaluj SQLAlchemy 2.1 lub nowszą z jej mssql-python zależnością:

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

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

• Stringi połączeń dla mssql-python
• Zarządzanie połączeniami za pomocą mssql-python
• Użyj mssql-python z FastAPI
• Używaj mssql-python z Flaskiem
• Dokumentacja dialektu SQLAlchemy MSSQL