Usa mssql-python con SQLAlchemy

SQLAlchemy è il toolkit Python ORM e database più utilizzato. A partire da SQLAlchemy 2.1.0b2, un dialetto integrato per il driver mssql-python ti permette di usare SQLAlchemy ORM e Core con Microsoft SQL e database SQL di Azure.

Prerequisiti

• Python 3.10 o versione successiva. SQLAlchemy 2.1 ha eliminato il supporto per Python 3.9 e precedenti.
• I pacchetti mssql-python e sqlalchemy (2.1.0b2 o versione successiva).

Gli esempi in questo articolo utilizzano il database di esempio AdventureWorksLT . Se non hai installato AdventureWorksLT, consulta i database di esempio di AdventureWorks.

Installa la versione preliminare

Poiché SQLAlchemy 2.1 è in beta, pip install sqlalchemy installa di default l'ultima versione stabile 2.0.x. Installa esplicitamente il pre-release:

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

Verificare la versione installata:

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

URL di connessione

Il dialetto mssql-python usa mssql+mssqlpython come schema URL. Il formato generale è:

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

Autenticazione SQL

Per l'autenticazione SQL, includere nome utente e password nell'URL di connessione:

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

Autenticazione Microsoft Entra

Per l'autenticazione con Microsoft Entra, usa un nome utente vuoto e il parametro di query authentication:

from sqlalchemy import create_engine

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

Creare URL a livello di codice

Da usare sqlalchemy.engine.URL.create per evitare la codifica manuale degli 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)

Definire modelli ORM

Usa la mappatura dichiarativa di SQLAlchemy per definire modelli che si mappano alle tabelle SQL di Microsoft.

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

Operazioni CRUD

I seguenti esempi mostrano come inserire, interrogare, aggiornare ed eliminare righe utilizzando la sessione ORM. Ogni esempio riutilizza new_id, il valore ProductID restituito quando viene inserita una riga. Per eseguire tutte e quattro le operazioni insieme, vedi l'esempio completo.

Creare una sessione

Crea una sessione per eseguire operazioni all'interno di una transazione:

from sqlalchemy.orm import Session

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

Per applicazioni che creano molte sessioni, si utilizza sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Inserire righe

Aggiungi un nuovo prodotto, esegui il commit della sessione e acquisisci il tag ProductID generato per i seguenti esempi:

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

Righe di query

Recupera una singola riga tramite chiave primaria, oppure usa select() per query filtrate:

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

Aggiornamento delle righe

Modifica un campo su una riga esistente e effettua un commit:

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

Elimina righe

Rimuovi una riga e fai il commit di:

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

Esempio completo

Le sezioni precedenti mostravano ogni pezzo separatamente. Questa sezione li combina in un unico script autonomo che puoi copiare, eseguire e ripetere.

Creare un file denominato crud.py e aggiungere il codice seguente. Sostituisci i dettagli create_engine di connessione con i tuoi (vedi URL di connessione):

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

Eseguire lo script:

python crud.py

Si vedono risultati simili ai seguenti:

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

Lo script elimina la riga che crea, così non colpisce i vincoli unici su Name e ProductNumber quando lo esegui di nuovo. Ogni esecuzione inserisce una nuova riga, quindi il valore di ProductID aumenta ogni volta.

Query principali

SQLAlchemy Core fornisce un'API di espressioni SQL di livello inferiore. Puoi usare Core con lo stesso motore e definizioni di tabelle, incluse le classi mappate con ORM.

from sqlalchemy import text

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

Usa costrutti a livello di tabella per la generazione di SQL con controllo dei tipi:

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

Pool di connessioni

SQLAlchemy gestisce di default un pool di connessioni. Regola le impostazioni del pool per il tuo carico di lavoro:

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

Utilizzo con framework web

SQLAlchemy è comunemente utilizzato come livello database per Flask e FastAPI. Il dialetto mssql-python funziona con qualsiasi framework che supporti SQLAlchemy.

I seguenti estratti mostrano il modello consigliato sessione-per-richiesta per ciascun framework. Sono frammenti illustrativi che assumono il engine modello e Product dalle sezioni precedenti, non app complete. Per applicazioni complete ed eseguibili, consulta gli articoli sull'integrazione FastAPI e sull'integrazione Flask .

Esempio di FastAPI

Usa una dipendenza dal generatore per fornire una sessione per richiesta:

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

Esempio di Flask

Usa un gestore di contesto per definire la sessione in base alla richiesta:

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

Migrazioni alambiche

Alembic gestisce le migrazioni degli schemi per i progetti SQLAlchemy e funziona con il dialetto mssql-python. La funzione autogenerate di Alembic confronta i tuoi modelli con il database live, quindi qualche passaggio in più impedisce che proponga modifiche alle tabelle che non gestisci.

Istituire Alembic

Installa Alembic e inizializza una directory di migrazione:

pip install alembic
alembic init migrations

In alembic.ini, imposta l'URL di connessione:

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

Indirizza Alembic ai tuoi modelli

Autogenerate ha bisogno dei metadati dei tuoi modelli. Inserisci i modelli gestiti da Alembic in un modulo importabile, come models.py. Poiché autogenerate propone di eliminare qualsiasi colonna omessa da un modello, definisci un modello che possiede completamente la sua tabella invece di riutilizzare il modello semplificato Product di questo articolo precedente:

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

In migrations/env.py, sostituisci target_metadata = None con il seguente codice. Importa i tuoi modelli e limita la generazione automatica agli schemi e alle tabelle che definiscono:

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

Passa include_name e include_schemas=True a context.configure sia in run_migrations_offline sia in run_migrations_online. L'impostazione include_schemas=True permette ad Alembic di vedere le tabelle in schemi non predefiniti come SalesLT:

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

Genera e applica una migrazione

Genera una migrazione dai tuoi modelli:

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

Alembic rileva la nuova tabella e scrive uno script di migrazione:

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

Il comando generato upgrade() crea la tabella, e downgrade() la rimuove:

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

Rivedi lo script e poi applica tutte le migrazioni in sospeso:

alembic upgrade head

Differenze rispetto al dialetto pyodbc

Se stai migrando da mssql+pyodbc, il dialetto mssql-python è simile perché entrambi i driver si basano sullo stesso framework ODBC. Differenze principali:

Limitazioni note

Il dialetto mssql-python per SQLAlchemy è in pre-release. Prima di usarlo in produzione, comprendi queste implicazioni:

• Modifiche API: Firme di metodo, tipi di eccezione e comportamento potrebbero cambiare prima della release stabile finale. Fissa sempre la tua versione SQLAlchemy a una build pre-release specifica (ad esempio, sqlalchemy==2.1.0b2) e testa accuratamente gli aggiornamenti.

• Test limitati: Il dialetto ha meno test comunitari rispetto al dialetto stabile mssql+pyodbc . Potresti incontrare casi limitali o funzionalità mancanti.

• Lacune di funzionalità: alcune funzionalità avanzate di ORM o Core potrebbero non funzionare. Consulta la documentazione dei dialetti MSSQL di SQLAlchemy e testa i tuoi casi d'uso prima di impegnarti in un progetto.

• Nessuna garanzia di supporto: Microsoft e SQLAlchemy forniscono un supporto al meglio dell'impegno, ma i problemi potrebbero non essere risolti prima della versione stabile.

Quando usare la versione preliminare:

• Ambienti di sviluppo e test
• Progetti di prova di concetto
• Migrazione da mssql+pyodbc se vuoi evitare la dipendenza da driver ODBC esterno
• Progetti in cui puoi rispondere a cambiamenti API ed eseguire test di regressione

Quando NON utilizzare la versione preliminare:

• Sistemi di produzione con requisiti di stabilità rigorosi
• Applicazioni legacy di lunga data in cui gli aggiornamenti delle dipendenze sono rari
• Carichi di lavoro aziendali critici in attesa che SQLAlchemy 2.1 raggiunga una versione GA stabile

Per lo stato dei dialetti pre-release più recenti e i problemi noti, controlla il repository GitHub di mssql-python.

Troubleshooting

"Nessun modulo chiamato 'sqlalchemy.dialects.mssql.mssqlpython'"

Questo errore significa che la versione installata di SQLAlchemy non include il dialetto mssql-python. Verifica di avere la versione 2.1.0b2 o successiva:

pip install "sqlalchemy>=2.1.0b2"

Errori di connessione

Se create_engine ha successo ma le query falliscono, verifica che i parametri di connessione funzionino direttamente con 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()

Se la connessione diretta funziona ma SQLAlchemy no, controlla eventuali problemi di codifica URL in caratteri speciali all'interno della password o del nome del server.

Contenuti correlati

• Stringhe di connessione mssql-python
• Gestione delle connessioni MSSQL-Python
• Integrazione FastAPI
• Integrazione con Flask
• Documentazione del dialetto MSSQL di SQLAlchemy