Megjegyzés
Az oldalhoz való hozzáféréshez engedély szükséges. Megpróbálhat bejelentkezni vagy módosítani a címtárat.
Az oldalhoz való hozzáféréshez engedély szükséges. Megpróbálhatja módosítani a címtárat.
Az SQLAlchemy a legelterjedtebb Python ORM és adatbázis eszköztár. Az SQLAlchemy 2.1.0b2-vel kezdve, egy beépített dialektus az mssql-python driverhez lehetővé teszi SQLAlchemy ORM és Core használatát Microsoft SQL és Azure SQL Database segítségével.
Important
Az mssql-python dialektust az SQLAlchemy 2.1.0b2-ben adták hozzá (2026. április 16-án jelent meg). Az SQLAlchemy 2.1 jelenleg előzetes kiadású sorozat, és nem ajánlott gyártási használatra. Mielőtt frissítené az SQLAlchemy 2.0-ról, értsd meg:
- Az API-k megváltozhatnak a végleges stabil kiadás (2.1 GA) előtt
- Teszteld alaposan a munkaterhelésedet a telepítés előtt
- Használja a stabil SQLAlchemy 2.0.x verziót éles rendszerekben, amíg a 2.1 általánosan elérhetővé nem válik
-
Rögzítsd a függőségedet egy adott verzióhoz (például
sqlalchemy==2.1.0b2) a verziótartományok helyett
Lásd az Ismert Korlátozások szekciót a részletekért arról, mikor érdemes használni az előzetes kiadású verziókat.
Prerequisites
- Python 3.10 vagy újabb verzió. Az SQLAlchemy 2.1 megszüntette a Python 3.9 és korábbi verziók támogatását.
- Az
mssql-pythonandsqlalchemycsomagok (2.1.0b2 vagy újabb).
A cikkben szereplő példák az AdventureWorksLT mintaadatbázist használják. Ha nincs telepítve az AdventureWorksLT, nézd meg az AdventureWorks mintaadatbázisokat.
Telepítsd az előkiadást
Mivel az SQLAlchemy 2.1 béta verzióban van, pip install sqlalchemy alapértelmezés szerint telepíti a legfrissebb stabil 2.0.x kiadást. Telepítsd kifejezetten az előzetes kiadást:
pip install mssql-python "sqlalchemy>=2.1.0b2"
Ellenőrizze a telepített verziót:
import sqlalchemy
print(sqlalchemy.__version__) # Should show 2.1.0b2 or later
Kapcsolati URL-ek
Az mssql-python dialektust mssql+mssqlpython használja URL sémaként. Az általános formátum a következő:
mssql+mssqlpython://<username>:<password>@<host>:<port>/<database>
SQL-hitelesítés
SQL hitelesítéshez a felhasználónevet és jelszót a kapcsolati URL-ben foglaljuk be:
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-hitelesítés
Microsoft Entra hitelesítéshez használj üres felhasználónevet és a authentication lekérdezési paramétert:
from sqlalchemy import create_engine
engine = create_engine(
"mssql+mssqlpython://@<server>.database.windows.net/<database>"
"?authentication=ActiveDirectoryDefault&encrypt=yes"
)
Megjegyzés:
ActiveDirectoryDefault használ DefaultAzureCredential, amely több hitelesítésszolgáltatót próbál egymás után. Az első kapcsolat lassú lehet, mert az SDK végigjárja a láncot, amíg meg nem talál egy működő szolgáltatót. A termelésben, ha tudod, melyik hitelesítéstípust használja a környezeted, közvetlenül megadd (például ActiveDirectoryMSI menedzselt identitásnál), hogy elkerüld a láncos sétát. További információ: Microsoft Entra-hitelesítés.
URL-ek programozott építése
A manuális URL-kódolás elkerüléséhez használja a(z) sqlalchemy.engine.URL.create elemet:
from sqlalchemy.engine import URL
url = URL.create(
"mssql+mssqlpython",
username="dbuser",
password="<password>",
host="localhost",
port=1433,
database="<database>",
)
engine = create_engine(url)
Definiáljuk az ORM modelleket
Használd az SQLAlchemy deklaratív leképezését olyan modellek definiálásához, amelyek a Microsoft SQL táblákhoz hasonlítottak.
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()
)
Jótanács
A Microsoft SQL az oszlopok automatikus növelésére használjaIDENTITY.
Az SQLAlchemy ezt automatikusan leképezi egész számú elsődleges kulcsoszlopokhoz. A fentebb bemutatott explicit Identity() használata nem kötelező, hacsak nem kell megadnia a kezdő- és növekményértékeket.
CRUD műveletek
Az alábbi példák bemutatják, hogyan lehet sorokat hozzáadni, lekérdezni, frissíteni és törölni az ORM ülés használatával. Minden példa újra felhasználja a new_id elemet, vagyis a ProductID értéket, amelyet egy sor beszúrásakor kap. A négy művelet együttes futtatásához lásd a teljes példát.
Munkamenet létrehozása
Hozz létre egy ülést a tranzakción belüli műveletek végrehajtásához:
from sqlalchemy.orm import Session
with Session(engine) as session:
# Use session for queries and modifications
pass
Olyan alkalmazásokhoz, amelyek sok munkamenetet hoznak létre, használd sessionmaker:
from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(bind=engine)
Sorok beszúrása
Új termék hozzáadása, a munkamenet elköteleződése, és a generált ProductID adatok rögzítése az alábbi példákhoz:
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}")
Megjegyzés:
A(z) SalesLT.Product esetében mind a Name, mind a ProductNumber egyedi megkötésekkel rendelkezik. Ha ezt a betétet többször futtatod, először változtasd meg ezeket az értékeket, vagy töröld az előző sort.
A teljes példa törli a létrehozott sort, így ismételten futhat.
Lekérdezéssorok
Egyetlen sor lekérése elsődleges kulcs alapján, vagy szűrt lekérdezésekhez való felhasználás 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}")
Sorok frissítése
Módosítsd egy meglévő sor mezőjét, és véglegesítsd:
with Session(engine) as session:
product = session.get(Product, new_id)
if product:
product.list_price = Decimal("1349.99")
session.commit()
Sorok törlése
Távolíts el egy sort, és véglegesítsd:
with Session(engine) as session:
product = session.get(Product, new_id)
if product:
session.delete(product)
session.commit()
Teljes példa
Az előző részek minden darabot külön mutattak be. Ez a rész egyesíti őket egy önálló szkriptté, amit lemásolhatsz, futtathatsz és újra futtathatsz.
Hozzon létre egy elnevezett crud.py fájlt, és adja hozzá a következő kódot. Cseréld ki a kapcsolati adatokat create_engine a sajátoddal (lásd Kapcsolati URL-ek):
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}")
Futtassa a szkriptet:
python crud.py
Hasonló kimenetet látsz, mint az alábbi:
Inserted ProductID: 1019
Read: Classic Road Bike costs $1299.9900
Updated price to $1349.9900
Deleted ProductID: 1019
A szkript törli az általa létrehozott sort, így amikor újra futtatja, nem ütközik a Name és ProductNumber egyediségi megkötéseibe. Minden egyes futtatás új sort szúr be, így a ProductID minden alkalommal növekszik.
Maglekérdezések
SQLAlchemy A Core alacsonyabb szintű SQL kifejezési API-t biztosít. Használhatod a Core-t ugyanazzal a motorral és tábladefinícióval, beleértve az ORM-térképes osztályokat is.
from sqlalchemy import text
with engine.connect() as conn:
result = conn.execute(text("SELECT @@VERSION"))
print(result.scalar())
Használjon táblázatszintű konstrukciókat típusbiztonsági SQL generáláshoz:
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()
Megjegyzés:
Amikor olyan egyedileg leképezett oszlopokat választ ki, amelyeknek az adatbázisbeli neve eltér az attribútum nevétől (például a(z) Product.name a(z) Name oszlopra van leképezve), a Core sorai az adatbázisoszlop neve alapján vannak kulcsolva. Adja hozzá a(z) .label("name") elemet, hogy az értéket row.name-ként érje el row.Name helyett.
Kapcsolatmegosztás
Az SQLAlchemy alapértelmezés szerint egy kapcsolati poolt kezel. Hangold a pool beállításait a munkaterhelésedhez:
engine = create_engine(
"mssql+mssqlpython://dbuser:<password>@localhost/<database>",
pool_size=10,
max_overflow=20,
pool_timeout=30,
pool_recycle=3600,
)
| Paraméter | Leírás |
|---|---|
pool_size |
A nyitva tartandó kapcsolatok száma (alapértelmezett: 5). |
max_overflow |
A(z) pool_size feletti kapcsolatok engedélyezettek (alapértelmezett: 10). |
pool_timeout |
Másodpercek a kapcsolatra várni, mielőtt hiba keletkezik (alapértelmezett: 30). |
pool_recycle |
Másodpercekkel később újrahasznosítják a kapcsolatot (alapértelmezett: -1, kikapcsolva). Állítsd be ezt az értéket, ha az adatbázisod lezárja az üres kapcsolatokat. |
Használat webes keretrendszerekkel
Az SQLAlchemy gyakran használják adatbázis-rétegként a Flask és a FastAPI számára. Az mssql-python dialektus bármely olyan keretrendszerrel működik, amely támogatja az SQLAlchemy-t.
A következő részletek bemutatják az ajánlott session-per request mintát minden keretrendszerhez. Ezek szemléltető részletek, amelyek a korábbi szakaszokban ismertetett engine és Product modellt feltételezik, nem teljes alkalmazások. Teljes, futtatható alkalmazásokért lásd a FastAPI integrációs és Flask integrációs cikkeket.
FastAPI példa
Használj generátoros függőséget, hogy minden kéréshez egy munkamenetet biztosíts:
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)}
Flaska példa
Használj kontextuskezelőt, hogy a munkamenetet a kéréshez igazítsd:
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 vándorlások
Alembic kezeli a SQLAlchemy-projektek sémamigrációit, és az mssql-python dialektussal működik. Az Alembic automatikus generáló funkciója összehasonlítja a modelljeidet az élő adatbázissal, így néhány plusz lépés megakadályozza, hogy olyan táblázatokban javasoljon változtatásokat, amelyeket nem kezelsz.
Alembic létrehozása
Telepítsd az Alembic-et és inicializáld a migrációs könyvtárat:
pip install alembic
alembic init migrations
A ben alembic.ini, állítsd be a kapcsolati URL-t:
sqlalchemy.url = mssql+mssqlpython://dbuser:<password>@localhost/<database>
Irányítsd az Alembicet a modelljeidre
Az Autogeneration-nak szüksége van a modelljeid metaadataira. Tedd az Alembic által kezelt modelleket egy importálható modulba, például models.py. Mivel az autogenerálás azt javasolja, hogy bármely oszlopot elhagyjon, amelyet egy modell kihagy, definiáljunk egy olyan modellt, amely teljes mértékben birtokolja a tábláját, ahelyett, hogy újra használnád a cikkben korábban említett egyszerűsített Product modellt:
# 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
Alapértelmezés szerint az automatikus generálás az adatbázis minden olyan tábláját, amely nem szerepel a(z) target_metadata elemben, eltávolítottnak tekinti, és drop_table utasítást hoz létre hozzá. Egy meglévő adatbázis, például az AdventureWorksLT esetében ez a művelet több tucat táblát is törölhet. Adj egy include_name szűrőt, hogy az Alembic csak azokat a táblázatokat kezelje, amiket a modelljeid definiálnak, és mindig nézze át a generált szkriptet, mielőtt alkalmaznád.
A(z) migrations/env.py elemben cseréld le a(z) target_metadata = None elemet a következő kódra. Importálja a modelleket, és az automatikus generálást az általuk meghatározott sémákra és táblákra korlátozza:
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
Add át a(z) include_name és include_schemas=True elemeket a(z) context.configure számára mind a(z) run_migrations_offline, mind a(z) run_migrations_online esetében. A include_schemas=True beállítás lehetővé teszi, hogy az Alembic nem alapértelmezett sémákban is lásson táblázatokat, például SalesLT:
context.configure(
connection=connection,
target_metadata=target_metadata,
include_name=include_name,
include_schemas=True,
)
Generálj és alkalmazz migrációt
Generálj migrációt a modelljeidből:
alembic revision --autogenerate -m "add product review table"
Az Alembic felismeri az új táblát, és migrációs szkriptet ír:
INFO [alembic.autogenerate.compare.tables] Detected added table 'SalesLT.ProductReview'
Generating .../versions/xxxx_add_product_review_table.py ... done
A generált upgrade() létrehozza a táblázatot, majd downgrade() eldobja:
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")
Nézd át a szkriptet, majd alkalmazd az összes függőben lévő migrációt:
alembic upgrade head
Különbségek a pyodbc dialektustól
Ha a(z) mssql+pyodbc rendszerről migrálsz, az mssql-python dialektus hasonló lesz, mivel mindkét illesztőprogram ugyanarra az ODBC keretrendszerre épül. Főbb különbségek:
| Téma | mssql+pyodbc |
mssql+mssqlpython |
|---|---|---|
| ODBC-illesztőprogram telepítése | Külön ODBC illevezetőt igényel (például ODBC Driver 18 Microsoft SQL-hez). | Az illesztőprogram mellékelve van. Nem kell külön ODBC driver. |
| Kapcsolat URL | mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server |
mssql+mssqlpython://user:pass@host/db |
fast_executemany |
Támogatva a következőn keresztül: create_engine(..., fast_executemany=True). |
Nem alkalmazható. A meghajtó belső szakaszteljesítményt kezeli. |
| Availability | Stabil, az SQLAlchemy-be van beépítve az 1.x óta. | Előzetes kiadás (SQLAlchemy 2.1.0b2+). |
Ismert korlátozások
Az mssql-python dialektusa SQLAlchemy számára előzetes kiadásban van. Mielőtt gyártásban használnánk, értsd meg ezeket a következményeket:
API változások: A metódusaláírások, kivételtípusok és viselkedés változhatnak a végleges stabil kiadás előtt. Mindig rögzítsd az SQLAlchemy verziódat egy adott előzetes verzióhoz (például),
sqlalchemy==2.1.0b2és alaposan teszteld a frissítéseket.Korlátozott tesztelés: A dialektusban kevesebb közösségi tesztelés van, mint a stabil
mssql+pyodbcdialektus. Előfordulhat, hogy szélső esetek vagy hiányzó funkciók is előfordulhatnak.Funkcióhiányok: Néhány fejlett ORM vagy Core funkció nem feltétlenül működik. Nézd meg az SQLAlchemy MSSQL dialektus dokumentációt , és teszteld a felhasználási eseteidet, mielőtt elköteleződnél egy projekthez.
Nincs támogatási garancia: a Microsoft és az SQLAlchemy a lehető legjobb támogatást nyújtja, de a problémák nem feltétlenül oldódnak meg a stabil kiadás előtt.
Mikor érdemes használni az előzetes kiadást:
- Fejlesztési és tesztelési környezetek
- Koncepció-bizonyítási projektek
- Áttérés erről:
mssql+pyodbc, ha el szeretné kerülni a külső ODBC-illesztőprogram-függőséget - Olyan projektek, ahol reagálhatsz API-változásokra és regressziós tesztelést végezhetsz
Mikor NE használd az előzetes kiadást:
- Szigorú stabilitási követelményeket igénylő gyártási rendszerek
- Többéves örökségi alkalmazások, ahol a függőségi frissítések ritkák
- Kritikus üzleti munkaterhelések, amíg az SQLAlchemy 2.1 stabil GA-t nem éri el
A legfrissebb kiadás előtti dialektus állapotért és ismert problémákért nézd meg az mssql-python GitHub repozitóriumot.
Hibaelhárítás
Nincs „sqlalchemy.dialects.mssql.mssqlpython” néven modul.
Ez a hiba azt jelenti, hogy a telepített SQLAlchemy verzió nem tartalmazza az mssql-python dialektust. Ellenőrizd, hogy 2.1.0b2 vagy újabb verziód van:
pip install "sqlalchemy>=2.1.0b2"
Csatlakozási hibák
Ha create_engine sikerül, de a lekérdezések kudarcosak, ellenőrizd, hogy a kapcsolati paramétereid közvetlenül az mssql-pythonnal működnek-e:
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()
Ha a közvetlen kapcsolat működik, de a SQLAlchemy nem, ellenőrizd, hogy nincs-e URL-kódolási probléma a jelszóban vagy a szervernévben szereplő speciális karakterekkel.