SQLAlchemy ile mssql-python kullanın

SQLAlchemy, en yaygın kullanılan Python ORM ve veritabanı araç setidir. SQLAlchemy 2.1.0b2 ile başlayarak, mssql-python sürücüsü için yerleşik bir lehçe sayesinde SQLAlchemy ORM ve Core ile Microsoft SQL ve Azure SQL Veritabanı ile kullanılabiliyor.

Important

mssql-python lehçesi, SQLAlchemy 2.1.0b2'de eklendi (16 Nisan 2026'da yayımlandı). SQLAlchemy 2.1 şu anda ön sürüm serisidir ve üretim kullanımı için önerilmez. SQLAlchemy 2.0'dan yükseltme yapmadan önce şunu anlayın:

  • API'ler, nihai stabil sürümden (2.1 GA) önce değişebilir.
  • Görevden önce iş yükünüzü iyice test edin
  • 2.1 GA seviyesine ulaşana kadar üretim sistemleri için stabil SQLAlchemy 2.0.x kullanın
  • Bağımlılığınızı belirli bir sürüme (örneğin, sqlalchemy==2.1.0b2) sabitleyin, sürüm aralıkları kullanmak yerine

Ön sürüm sürümlerinin ne zaman kullanılacağına dair detaylar için Bilinen Sınırlamalar bölümüne bakınız.

Prerequisites

  • Python 3.10 veya üzeri. SQLAlchemy 2.1, Python 3.9 ve daha önceki sürümleri desteklemeyi bıraktı.
  • ve mssql-pythonsqlalchemy paketleri (2.1.0b2 veya daha sonra).

Bu makaledeki örnekler AdventureWorksLT örnek veritabanını kullanır. Eğer AdventureWorksLT'niz yüklü değilse, AdventureWorks örnek veritabanlarına bakabilirsiniz.

Ön sürümü kur

SQLAlchemy 2.1 beta sürümünde olduğu için pip install sqlalchemy varsayılan olarak en son kararlı 2.0.x sürümünü yükler. Ön sürümü açıkça kurun:

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

Yüklü sürümü doğrulayın:

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

Bağlantı URL'leri

mssql-python lehçesi URL şeması olarak kullanır mssql+mssqlpython . Genel biçim:

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

SQL kimlik doğrulaması

SQL kimlik doğrulaması için, bağlantı URL'sine kullanıcı adı ve şifreyi ekleyin:

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 kimlik doğrulama

Microsoft Entra kimlik doğrulaması için boş bir kullanıcı adı ve authentication sorgu parametresi kullanın:

from sqlalchemy import create_engine

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

Note

ActiveDirectoryDefault DefaultAzureCredential, kullanır ve bu da birden fazla kimlik doğrulama sağlayıcısını sırayla dener. İlk bağlantı yavaş olabilir çünkü SDK çalışan bir sağlayıcı bulana kadar zincirde yürür. Üretimde, ortamınızın hangi kimlik bilgisi türünü kullandığını biliyorsanız, zincirde dolaşmayı önlemek için bunu doğrudan belirtin (örneğin, yönetilen kimlik için ActiveDirectoryMSI). Daha fazla bilgi için bkz . Microsoft Entra kimlik doğrulaması.

URL'leri programatik olarak oluştur

Manuel URL kodlamasından kaçınmak için kullanın sqlalchemy.engine.URL.create :

from sqlalchemy.engine import URL

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

ORM modellerini tanımlayın

Microsoft SQL tablolarına eşleyen modelleri tanımlamak için SQLAlchemy'nin bildirmesel eşlemesini kullanın.

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, otomatik olarak sütun artırma için kullanırIDENTITY. SQLAlchemy bunu tam sayı birincil anahtar sütunları için otomatik olarak eşler. Yukarıda gösterilen açık Identity() ifade isteğe bağlıdır, başlangıç ve artış değerlerini kontrol etmeniz gerekmedikçe.

CRUD operasyonları

Aşağıdaki örnekler, ORM oturumu kullanılarak satır ekleme, sorgulama, güncelleme ve silme yöntemlerini gösterir. Her örnek, bir satır eklediğinizde döndürülen ProductID olan new_id öğesini yeniden kullanır. Dört işlemi birlikte çalıştırmak için tam örneği görün.

Oturum oluşturma

Bir işlem içinde işlemleri yürütmek için bir oturum oluşturun:

from sqlalchemy.orm import Session

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

Birçok oturum oluşturan uygulamalar için şunları kullanın sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Satır ekleme

Yeni bir ürün ekleyin, oturumu kaydedin ve aşağıdaki örnekler için oluşturulan ProductID öğesini yakalayın:

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

SalesLT.Product içinde, hem Name hem de ProductNumber benzersiz kısıtlamalara sahiptir. Bu eklemeyi birden fazla kez çalıştırırsanız, bu değerleri değiştirin veya önce önceki satırı silin. Tam örnek, oluşturduğu satırı siler, böylece tekrar tekrar çalıştırabilir.

Sorgu satırları

Birincil anahtarla tek bir satırı alın veya filtrelenmiş sorgular için kullanın select() :

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

Satırları güncelleştirme

Mevcut bir satırdaki bir alanı değiştirin ve commit yapın:

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

Satırları silme

Bir satırı kaldır ve commit et:

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

Tam örnek

Önceki bölümler her parçayı ayrı ayrı gösteriyordu. Bu bölüm, bunları kopyalayıp çalıştırıp tekrar çalıştırabileceğiniz tek bir bağımsız betikte birleştiriyor.

crud.py adlı bir dosya oluşturun ve aşağıdaki kodu ekleyin. Bağlantı detaylarını create_engine kendi bilgilerinizle değiştirin ( bkz. Bağlantı URL'leri):

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

Betiği Çalıştırın:

python crud.py

Aşağıdaki gibi bir çıktı görüyorsunuz:

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

Komut dosyası, oluşturduğu satırı siler; böylece yeniden çalıştırdığınızda Name ve ProductNumber üzerindeki benzersiz kısıtlamaları ihlal etmez. Her çalıştırma yeni bir satır ekler, bu yüzden ProductID her seferinde artar.

Temel sorgular

SQLAlchemy Core, daha düşük seviyeli bir SQL ifade API'si sağlar. Core'u aynı motor ve tablo tanımlarında, ORM haritalı sınıflar da dahil olmak üzere kullanabilirsiniz.

from sqlalchemy import text

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

Tip güvenli SQL üretimi için tablo düzeyinde yapılar kullanın:

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

Veritabanındaki adı öznitelik adından farklı olan tek tek eşlenmiş sütunları seçtiğinizde (örneğin, Product.name, Name sütununa eşlenir), Core satırları veritabanı sütun adına göre anahtarlanır. Değere row.Name yerine row.name olarak erişmek için .label("name") ekleyin.

Bağlantı havuzlama

SQLAlchemy varsayılan olarak bir bağlantı havuzunu yönetir. İş yükünüz için havuz ayarlarını ayarlayın:

engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost/<database>",
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=3600,
)
Parametre Açıklama
pool_size Açık tutulacak bağlantı sayısı (varsayılan: 5).
max_overflow pool_size üzerinde izin verilen bağlantılar (varsayılan: 10).
pool_timeout Hata oluşturmadan önce bağlantı için beklenecek saniye sayısı (varsayılan: 30).
pool_recycle Saniyeler sonra bağlantı geri dönüştürülür (varsayılan: -1, devre dışı bırakılır). Veritabanınız boşta bağlantıları kapatıyorsa bu değeri ayarlayın.

Web çerçeveleriyle kullanım

SQLAlchemy , genellikle Flask ve FastAPI için veritabanı katmanı olarak kullanılır. mssql-python lehçesi, SQLAlchemy'yi destekleyen herhangi bir çerçeveyle çalışır.

Aşağıdaki kod parçacıkları, her çerçeve için önerilen istek başına bir oturum düzenini gösterir. Bunlar, önceki bölümlerdeki engine ve Product modelini varsayan açıklayıcı parçalardır; tam uygulamalar değildir. Tam ve çalıştırılabilir uygulamalar için FastAPI entegrasyonu ve Flask entegrasyon makalelerine bakınız.

FastAPI örneği

İstek başına bir oturum sağlamak için bir jeneratör bağımlılığı kullanın:

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

Matla örneği

Oturumu isteğe bağlamak için bir bağlam yöneticisi kullanın:

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

Alembik göçler

Alembic , SQLAlchemy projeleri için şema geçişlerini yönetir ve mssql-python lehçesiyle çalışır. Alembic'in otomatik oluşturma özelliği, modellerinizi canlı veritabanıyla karşılaştırıyor, bu yüzden birkaç ekstra adım, yönetemediğiniz tablolara değişiklik önermesini engelliyor.

Alembic'i kur

Alembic'i kur ve bir göç dizinini başlat:

pip install alembic
alembic init migrations

Bağlantı URL'sini ayarlayın alembic.ini:

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

Alembic'i modellerinize yöneltin

Otomatik oluşturma modellerinizin meta verilerine ihtiyaç duyar. Alembic'in yönettiği modelleri aktarılabilir bir modüle koyun, örneğin models.py. Otomatik üretim, bir modelde yer almayan herhangi bir sütunun kaldırılmasını önerdiği için, bu makalenin önceki bölümünde kullanılan basitleştirilmiş Product modelini yeniden kullanmak yerine tablosunun tamamını kapsayan bir model tanımlayın:

# 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

Varsayılan olarak, otomatik oluşturma veritabanındaki target_metadata içinde olmayan her tabloyu kaldırılmış kabul eder ve bunun için drop_table üretir. AdventureWorksLT gibi mevcut bir veritabanına karşı, bu işlem onlarca tabloyu düşürebilir. Alembic'in sadece modellerinizin tanımladığı tabloları yönetmesi için bir include_name filtre ekleyin ve uygulamadan önce oluşturulan betikleri her zaman gözden geçirin.

migrations/env.py içinde, target_metadata = None öğesini aşağıdaki kodla değiştirin. Modellerinizi içe aktarır ve otomatik oluşturmayı, tanımladıkları şema ve tablolarla sınırlar:

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

Hem run_migrations_offline hem de run_migrations_online içinde include_name ve include_schemas=True değerlerini context.configure'ye iletin. Bu include_schemas=True ayar, Alembic'in varsayılan olmayan şemalarda tabloları görmesini sağlıyor, örneğin SalesLT:

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

Bir göç oluşturun ve uygulayın

Modellerinizden bir migrasyon oluşturun:

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

Alembic yeni tabloyu algılar ve bir göç betiği yazar:

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

Oluşturulan upgrade() tabloyu oluşturur, downgrade() ise tabloyu siler:

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

Script'i inceleyin ve ardından tüm bekleyen göçleri uygulayın:

alembic upgrade head

pyodbc diyalektinden farklar

Eğer mssql+pyodbc sürücüsünden geçiş yapıyorsanız, mssql-python diyalekti benzerdir; çünkü her iki sürücü de aynı ODBC altyapısına dayanır. Önemli farklar:

Konu mssql+pyodbc mssql+mssqlpython
ODBC sürücü kurulumu Ayrı bir ODBC sürücüsü gerektirir (örneğin, Microsoft SQL için ODBC Driver 18). Sürücü dahildir. Ayrı bir ODBC sürücüsü gerekmiyor.
Bağlantı URL'si mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server mssql+mssqlpython://user:pass@host/db
fast_executemany create_engine(..., fast_executemany=True) aracılığıyla desteklenir. Uygulanamaz. Sürücü, toplu performansı dahili olarak yönetir.
Availability Stabil, 1.x'ten beri SQLAlchemy'ye dahil edilmiştir. Ön sürüm (SQLAlchemy 2.1.0b2+).

Bilinen Sınırlamalar

SQLAlchemy için mssql-python lehçesi ön sürüm aşamasındadır. Üretimde kullanmadan önce şu sonuçları anlayın:

  • API Değişiklikleri: Metod imzaları, istisna türleri ve davranışlar son kararlı sürümden önce değişebilir. Her zaman SQLAlchemy sürümünüzü belirli bir ön sürüm sürümüne sabitleyin (örneğin, sqlalchemy==2.1.0b2) ve yükseltmeleri iyice test edin.

  • Sınırlı Testler: Lehçe, kararlı mssql+pyodbc lehçeye göre daha az topluluk testi içerir. İstisnai durumlarla veya eksik işlevlerle karşılaşabilirsiniz.

  • Özellik Boşlukları: Bazı gelişmiş ORM veya Core özellikleri çalışmayabilir. SQLAlchemy MSSQL lehçe dokümantasyonuna bakın ve bir projeye başlamadan önce kullanım durumlarınızı test edin.

  • Destek Garantisi Yok: Microsoft ve SQLAlchemy en iyi şekilde destek sağlar, ancak sorunlar stabil sürümden önce çözülmeyebilir.

Ön Sürümün Ne Zaman Kullanılacağı:

  • Geliştirme ve test ortamları
  • Kavram kanıtı projeleri
  • Harici ODBC sürücüsü bağımlılığından kaçınmak istiyorsanız mssql+pyodbc kaynağından geçiş
  • API değişikliklerine yanıt verebileceğiniz ve regresyon testi yapabileceğiniz projeler

Ön Sürüm Kullanımını Ne Zaman Kullanmamalı:

  • Sıkı stabilite gereksinimlerine sahip üretim sistemleri
  • Bağımlılık güncellemelerinin nadir olduğu çok yıllık eski uygulamalar
  • SQLAlchemy 2.1 stabil GA'ya ulaşana kadar kritik iş yükleri

En son ön sürüm lehçe durumu ve bilinen sorunlar için mssql-python GitHub deposunu kontrol edin.

Sorun giderme

"'sqlalchemy.dialects.mssql.mssqlpython' adlı modül yok"

Bu hata, yüklenmiş SQLAlchemy sürümünüzün mssql-python lehçesini içermediği anlamına gelir. 2.1.0b2 veya daha yeni bir sürümüne sahip olduğunuzu doğrulayın:

pip install "sqlalchemy>=2.1.0b2"

Bağlantı hataları

Eğer create_engine başarılı olursa ama sorgular başarısız olursa, bağlantı parametrelerinizin doğrudan mssql-python ile çalıştığını doğrulayın:

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

Doğrudan bağlantı çalışıyorsa ama SQLAlchemy çalışmıyorsa, şifreniz veya sunucu adınızdaki özel karakterlerle URL kodlama sorunlarını kontrol edin.

İlgili içerik

  • Last updated on