Usa mssql-python con SQLAlchemy

SQLAlchemy è il toolkit Python ORM e database più utilizzato. SQLAlchemy 2.1 include un dialetto integrato per il mssql-python driver che puoi usare per lavorare con SQLAlchemy ORM e Core con Microsoft SQL e database SQL di Azure.

Prerequisiti

Python 3.11 o versione successiva. SQLAlchemy 2.1 ha eliminato il supporto per Python 3.10 e precedenti.
I pacchetti mssql-python e sqlalchemy (versione 2.1 o successive).

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

Installa SQLAlchemy e mssql-python

Usa la dipendenza opzionale di mssql-pythonSQLAlchemy per installare entrambi i pacchetti:

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

SQLAlchemy definisce l'extra mssql-python con mssql-python>=1.9.0. È supportato anche il formato esistente in pacchetti separati:

pip install "sqlalchemy>=2.1" mssql-python

Verificare la versione installata:

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

Note

ActiveDirectoryDefault usa DefaultAzureCredential, che prova più fornitori di credenziali in sequenza. La prima connessione può essere lenta perché l'SDK percorre la catena finché non trova un fornitore funzionante. In produzione, se sai quale tipo di credenziale utilizza il tuo ambiente, specificalo direttamente (ad esempio, ActiveDirectoryMSI per l'identità gestita) per evitare il chain walk. Per altre informazioni, vedere Autenticazione di Microsoft Entra.

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

Tip

Microsoft SQL utilizza IDENTITY l'incremento automatico delle colonne. SQLAlchemy mappa automaticamente questo parametro per le colonne chiave primarie intere. L'esplicito Identity() mostrato sopra è opzionale, a meno che tu non debba controllare i valori di inizio e incremento.

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

Note

In SalesLT.Product, sia Name che ProductNumber hanno vincoli unici. Se esegui questo insert più di una volta, cambia questi valori o elimina prima la riga precedente. L'esempio completo elimina la riga che crea, così da poterlo eseguire ripetutamente.

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

Note

Quando selezioni colonne mappate singole il cui nome del database differisce dal nome dell'attributo (ad esempio, Product.name mappa alla Name colonna), le righe Core sono codificate dal nome della colonna del database. Aggiungi .label("name") per accedere al valore come row.name invece di row.Name.

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

Parametro	Description
`pool_size`	Numero di connessioni da mantenere aperte (predefinito: 5).
`max_overflow`	Connessioni consentite oltre `pool_size` (predefinito: 10).
`pool_timeout`	Pochi secondi per aspettare una connessione prima di generare un errore (predefinito: 30).
`pool_recycle`	Pochi secondi dopo di che una connessione viene riciclata (predefinito: -1, disabilitato). Imposta questo valore se il tuo database chiude le connessioni inattive.

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

Caution

Per impostazione predefinita, autogenerate considera rimossa ogni tabella nel database che non è in target_metadata e genera drop_table per essa. Contro un database esistente come AdventureWorksLT, quell'azione può far cadere decine di tabelle. Aggiungi un include_name filtro così Alembic gestisce solo le tabelle definite dai tuoi modelli, e rivedi sempre lo script generato prima di applicarlo.

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:

Topic	`mssql+pyodbc`	`mssql+mssqlpython`
Installazione del driver ODBC	Richiede un driver ODBC separato (ad esempio, ODBC Driver 18 per SQL Server).	Installato automaticamente come dipendenza da pacchetto. Non è necessaria l'installazione separata di driver ODBC.
URL connessione	`mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server`	`mssql+mssqlpython://user:pass@host/db`
`fast_executemany`	Supportato tramite `create_engine(..., fast_executemany=True)`.	Non applicabile. Il driver gestisce internamente le prestazioni dell'elaborazione batch.
Disponibilità	Stabile, inclusa in SQLAlchemy fin dalla versione 1.x.	Stabile, incluso in SQLAlchemy 2.1 o successivo.

Considerazioni sull'aggiornamento

SQLAlchemy 2.1 include cambiamenti comportamentali che potrebbero influenzare l'aggiornamento delle applicazioni rispetto a versioni precedenti:

SQLAlchemy 2.1 richiede Python 3.11 o successivo.
Aggiorna le applicazioni SQLAlchemy 1.x a SQLAlchemy 2.0 prima di passare alla 2.1.
Testa le applicazioni esistenti di SQLAlchemy 2.0 rispetto ai cambiamenti comportamentali descritti in What's New in SQLAlchemy 2.1?

Gli esempi in questo articolo utilizzano l'API sincrona di SQLAlchemy. Le applicazioni che utilizzano il supporto asyncio di SQLAlchemy devono installare l'extra asyncio perché SQLAlchemy 2.1 non installa più greenlet per impostazione predefinita.

Troubleshooting

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

Questo errore significa che la versione installata di SQLAlchemy non include il mssql-python dialetto. Installa SQLAlchemy 2.1 o versione successiva insieme alla relativa mssql-python dipendenza:

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

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.

Commenti e suggerimenti

Questa pagina è stata utile?

Last updated on 2026-09-25

Usa mssql-python con SQLAlchemy

Prerequisiti

Installa SQLAlchemy e mssql-python

URL di connessione

Autenticazione SQL

Autenticazione Microsoft Entra

Creare URL a livello di codice

Definire modelli ORM

Operazioni CRUD

Creare una sessione

Inserire righe

Righe di query

Aggiornamento delle righe

Elimina righe

Esempio completo

Query principali

Pool di connessioni

Utilizzo con framework web

Esempio di FastAPI

Esempio di Flask

Migrazioni alambiche

Istituire Alembic

Indirizza Alembic ai tuoi modelli

Genera e applica una migrazione

Differenze rispetto al dialetto pyodbc

Considerazioni sull'aggiornamento

Troubleshooting

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

Errori di connessione

Contenuti correlati

Commenti e suggerimenti

Risorse aggiuntive