Menggunakan mssql-python dengan SQLAlchemy

SQLAlchemy adalah toolkit Python ORM dan database yang paling banyak digunakan. Dimulai dengan SQLAlchemy 2.1.0b2, dialek bawaan untuk driver mssql-python memungkinkan Anda menggunakan SQLAlchemy ORM dan Core dengan Microsoft SQL dan Azure SQL Database.

Penting

Dialek mssql-python ditambahkan di SQLAlchemy 2.1.0b2 (dirilis 16 April 2026). SQLAlchemy 2.1 saat ini merupakan seri pra-rilis dan tidak direkomendasikan untuk penggunaan produksi. Sebelum memutakhirkan dari SQLAlchemy 2.0, pahami:

• API mungkin berubah sebelum rilis stabil akhir (2.1 GA)
• Uji beban kerja Anda secara menyeluruh sebelum penerapan
• Gunakan SQLAlchemy 2.0.x yang stabil untuk sistem produksi hingga 2.1 mencapai GA
• Sematkan dependensi Anda ke versi tertentu (misalnya, sqlalchemy==2.1.0b2) daripada menggunakan rentang versi

Lihat bagian Batasan yang Diketahui untuk detail tentang kapan harus menggunakan versi prarilis.

Prasyarat

• Python 3.10 atau yang lebih baru. SQLAlchemy 2.1 membatalkan dukungan untuk Python 3.9 dan yang lebih lama.
• Paket mssql-python dan sqlalchemy (2.1.0b2 atau lebih baru).

Contoh dalam artikel ini menggunakan database sampel AdventureWorksLT . Jika Anda belum menginstal AdventureWorksLT, lihat Database sampel AdventureWorks.

Instal versi prarilis

Karena SQLAlchemy 2.1 dalam versi beta, pip install sqlalchemy menginstal rilis stabil terbaru 2.0.x secara default. Instal pra-rilis secara eksplisit:

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

Verifikasi versi yang diinstal:

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

URL koneksi

Dialek mssql-python digunakan mssql+mssqlpython sebagai skema URL. Format umumnya adalah:

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

Autentikasi SQL

Untuk autentikasi SQL, sertakan nama pengguna dan kata sandi dalam URL koneksi:

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 otentikasi

Untuk autentikasi Microsoft Entra, gunakan nama pengguna kosong dan authentication parameter kueri:

from sqlalchemy import create_engine

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

Note

ActiveDirectoryDefault menggunakan DefaultAzureCredential, yang mencoba beberapa penyedia kredensial secara berurutan. Koneksi pertama bisa lambat karena SDK menelusuri rantai hingga menemukan penyedia yang berfungsi. Dalam lingkungan produksi, jika Anda mengetahui jenis kredensial yang digunakan oleh lingkungan Anda, tentukan secara langsung (misalnya, ActiveDirectoryMSI untuk identitas terkelola) agar terhindar dari penelusuran berantai. Untuk informasi selengkapnya, lihat Autentikasi Microsoft Entra.

Membuat URL secara terprogram

Gunakan sqlalchemy.engine.URL.create untuk menghindari pengodean URL manual:

from sqlalchemy.engine import URL

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

Tentukan model ORM

Gunakan pemetaan deklaratif SQLAlchemy untuk menentukan model yang dipetakan ke tabel 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 digunakan IDENTITY untuk kolom penambahan otomatis. SQLAlchemy memetakan ini secara otomatis untuk kolom kunci primer bilangan bulat. Eksplisit Identity() yang ditampilkan di atas bersifat opsional kecuali Anda perlu mengontrol nilai awal dan peningkatan.

Operasi CRUD

Contoh berikut menunjukkan cara menyisipkan, mengkueri, memperbarui, dan menghapus baris dengan menggunakan sesi ORM. Setiap contoh menggunakan kembali new_id, yaitu ProductID yang dikembalikan ketika Anda menyisipkan baris. Untuk menjalankan keempat operasi secara bersamaan, lihat contoh lengkapnya.

Membuat sesi

Buat sesi untuk menjalankan operasi dalam transaksi:

from sqlalchemy.orm import Session

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

Untuk aplikasi yang membuat banyak sesi, gunakan sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Sisipkan baris

Tambahkan produk baru, terapkan sesi, dan tangkap produk yang dihasilkan ProductID untuk contoh berikut:

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

Di SalesLT.Product, keduanya Name dan ProductNumber memiliki batasan unik. Jika Anda menjalankan sisipan ini lebih dari sekali, ubah nilai ini atau hapus baris sebelumnya terlebih dahulu. Contoh lengkap menghapus baris yang dibuatnya, sehingga dapat berjalan berulang kali.

Baris kueri

Ambil satu baris dengan kunci utama, atau gunakan select() untuk kueri yang difilter:

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

Memperbarui baris

Ubah bidang pada baris yang ada dan terapkan:

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

Menghapus baris

Hapus satu baris dan lakukan commit:

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

Contoh lengkap

Bagian sebelumnya menunjukkan setiap bagian secara terpisah. Bagian ini menggabungkannya menjadi satu skrip mandiri yang dapat Anda salin, jalankan, dan jalankan lagi.

Buat file bernama crud.py dan tambahkan kode berikut. Ganti detail koneksi dengan detail create_engine Anda sendiri (lihat URL koneksi):

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

Jalankan skrip:

python crud.py

Anda melihat output yang mirip dengan berikut ini:

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

Skrip menghapus baris yang dibuatnya, sehingga tidak mencapai batasan unik pada Name dan ProductNumber saat Anda menjalankannya lagi. Setiap eksekusi menyisipkan baris baru, sehingga ProductID meningkat setiap kali.

Kueri utama

SQLAlchemy Core menyediakan API ekspresi SQL tingkat rendah. Anda dapat menggunakan Core dengan definisi mesin dan tabel yang sama, termasuk kelas yang dipetakan ORM.

from sqlalchemy import text

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

Gunakan konstruksi tingkat tabel untuk pembuatan SQL yang aman untuk jenis:

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

Saat Anda memilih kolom yang dipetakan secara individual yang nama database-nya berbeda dari nama atribut (misalnya, Product.name dipetakan ke kolom Name), baris Core diidentifikasi berdasarkan nama kolom database. Tambahkan .label("name") untuk mengakses nilai sebagai row.name alih-alih row.Name.

Pemanfaatan koneksi

SQLAlchemy mengelola kumpulan koneksi secara default. Sesuaikan pengaturan pool sesuai beban kerja Anda:

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

Parameter	Deskripsi
pool_size	Jumlah koneksi yang harus tetap terbuka (default: 5).
max_overflow	Koneksi diizinkan melampaui pool_size (default: 10).
pool_timeout	Detik untuk menunggu koneksi sebelum memunculkan kesalahan (default: 30).
pool_recycle	Detik setelah koneksi didaur ulang (default: -1, dinonaktifkan). Tetapkan nilai ini jika database Anda menutup koneksi yang tidak aktif.

Gunakan dengan kerangka kerja web

SQLAlchemy biasanya digunakan sebagai lapisan database untuk Flask dan FastAPI. Dialek mssql-python bekerja dengan kerangka kerja apa pun yang mendukung SQLAlchemy.

Cuplikan berikut menunjukkan pola sesi per permintaan yang direkomendasikan untuk setiap kerangka kerja. Itu sekadar fragmen ilustratif yang mengasumsikan model engine dan Product yang dibahas pada bagian sebelumnya, bukan aplikasi yang lengkap. Untuk aplikasi lengkap yang dapat dijalankan, lihat artikel integrasi FastAPI dan integrasi Flask .

Contoh FastAPI

Gunakan dependensi generator untuk menyediakan sesi per permintaan:

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

Contoh Flask

Gunakan manajer konteks untuk membatasi sesi ke permintaan tersebut:

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

Migrasi Alembic

Alembic menangani migrasi skema untuk proyek SQLAlchemy , dan bekerja dengan dialek mssql-python. Fitur pembuatan otomatis Alembic membandingkan model Anda dengan database langsung, jadi beberapa langkah tambahan mencegahnya mengusulkan perubahan pada tabel yang tidak Anda kelola.

Siapkan Alembic

Instal Alembic dan inisialisasi direktori migrasi:

pip install alembic
alembic init migrations

Di alembic.ini, atur URL koneksi:

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

Arahkan Alembic ke model Anda

Pembuatan otomatis membutuhkan metadata model Anda. Tempatkan model yang dikelola Alembic dalam modul yang dapat diimpor, seperti models.py. Karena autogenerate mengusulkan untuk menghapus kolom apa pun yang dihilangkan oleh model, tentukan model yang sepenuhnya memiliki tabelnya daripada menggunakan kembali model yang disederhanakan Product dari sebelumnya dalam artikel ini:

# 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

Secara bawaan, autogenerate menganggap setiap tabel dalam database yang tidak ada di target_metadata sebagai telah dihapus dan menghasilkan drop_table untuk tabel tersebut. Pada basis data yang sudah ada seperti AdventureWorksLT, tindakan itu dapat menghapus lusinan tabel. Tambahkan filter include_name sehingga Alembic hanya mengelola tabel yang ditentukan model Anda, dan selalu tinjau skrip yang dihasilkan sebelum Anda menerapkannya.

Di migrations/env.py, ganti target_metadata = None dengan kode berikut. Ini mengimpor model Anda dan membatasi pembuatan otomatis ke skema dan tabel yang mereka tentukan:

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

Teruskan include_name dan include_schemas=True ke context.configure baik di run_migrations_offline maupun run_migrations_online. Pengaturan ini include_schemas=True memungkinkan Alembic melihat tabel dalam skema nondefault seperti SalesLT:

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

Membuat dan menerapkan migrasi

Buat migrasi dari model Anda:

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

Alembic mendeteksi tabel baru dan menulis skrip migrasi:

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

upgrade() yang dihasilkan membuat tabel, dan downgrade() menghapusnya:

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

Tinjau skrip, lalu terapkan semua migrasi yang tertunda:

alembic upgrade head

Perbedaan dari dialek pyodbc

Jika Anda bermigrasi dari mssql+pyodbc, dialek mssql-python serupa karena kedua driver didasarkan pada kerangka kerja ODBC yang sama. Perbedaan utama:

Topik	mssql+pyodbc	mssql+mssqlpython
Instalasi driver ODBC	Memerlukan driver ODBC terpisah (misalnya, Driver ODBC 18 untuk Microsoft SQL).	Driver disertakan. Tidak diperlukan driver ODBC terpisah.
URL Koneksi	mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server	mssql+mssqlpython://user:pass@host/db
fast_executemany	Didukung melalui create_engine(..., fast_executemany=True).	Tidak dapat diterapkan. driver menangani kinerja batch secara internal.
Availability	Stabil, termasuk dalam SQLAlchemy sejak 1.x.	Pra-rilis (SQLAlchemy 2.1.0b2+).

Batasan yang Diketahui

Dialek mssql-python untuk SQLAlchemy berada dalam pra-rilis. Sebelum digunakan dalam produksi, pahami implikasi ini:

• Perubahan API: Tanda tangan metode, jenis pengecualian, dan perilaku dapat berubah sebelum rilis stabil akhir. Selalu sematkan versi SQLAlchemy Anda ke build pra-rilis tertentu (misalnya, sqlalchemy==2.1.0b2) dan uji peningkatan secara menyeluruh.

• Pengujian Terbatas: Dialek ini memiliki pengujian komunitas yang lebih sedikit daripada dialek stabil mssql+pyodbc . Anda mungkin menemukan kasus tepi atau fitur yang hilang.

• Kesenjangan Fitur: Beberapa fitur ORM atau Inti tingkat lanjut mungkin tidak berfungsi. Lihat dokumentasi dialek SQLAlchemy MSSQL dan uji kasus penggunaan Anda sebelum berkomitmen pada proyek.

• Tidak Ada Jaminan Dukungan: Microsoft dan SQLAlchemy memberikan dukungan terbaik, tetapi masalah mungkin tidak terselesaikan sebelum rilis stabil.

Kapan Menggunakan Pra-Rilis:

• Lingkungan pengembangan dan pengujian
• Proyek pembuktian konsep
• Bermigrasi dari mssql+pyodbc jika Anda ingin menghindari dependensi driver ODBC eksternal
• Proyek tempat Anda dapat merespons perubahan API dan melakukan pengujian regresi

Kapan TIDAK Menggunakan Pra-Rilis:

• Sistem produksi dengan persyaratan stabilitas yang ketat
• Aplikasi warisan yang telah digunakan selama bertahun-tahun dan jarang mengalami pembaruan dependensi
• Beban kerja bisnis penting hingga SQLAlchemy 2.1 mencapai GA yang stabil

Untuk status dialek pra-rilis terbaru dan masalah yang diketahui, periksa repositori GitHub mssql-python.

Troubleshooting

"Tidak ada modul bernama 'sqlalchemy.dialects.mssql.mssqlpython'"

Kesalahan ini berarti versi SQLAlchemy yang Anda instal tidak menyertakan dialek mssql-python. Pastikan Anda memiliki 2.1.0b2 atau yang lebih baru:

pip install "sqlalchemy>=2.1.0b2"

Kegagalan koneksi

Jika create_engine berhasil tetapi kueri gagal, verifikasi parameter koneksi Anda berfungsi dengan mssql-python secara langsung:

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

Jika koneksi langsung berfungsi, tetapi SQLAlchemy tidak, periksa masalah pengkodean URL dalam karakter khusus dalam kata sandi atau nama server Anda.

Konten terkait

• String koneksi mssql-python
• Manajemen koneksi mssql-python
• Integrasi FastAPI
• Integrasi Flask
• Dokumentasi dialek MSSQL SQLAlchemy