Utilisez mssql-python avec SQLAlchemy

SQLAlchemy est la boîte à outils Python ORM et de bases de données la plus utilisée. À partir de SQLAlchemy 2.1.0b2, un dialecte intégré au pilote mssql-python vous permet d’utiliser SQLAlchemy ORM et Core avec Microsoft SQL et Azure SQL Database.

Important

Le dialecte mssql-python a été ajouté dans SQLAlchemy 2.1.0b2 (publié le 16 avril 2026). SQLAlchemy 2.1 est actuellement une série en pré-sortie et n’est pas recommandée pour une utilisation en production. Avant de passer de SQLAlchemy 2.0, comprenez :

Les API peuvent changer avant la version stable finale (2.1 GA)
Testez minutieusement votre charge de travail avant le déploiement
Utilisez la version stable de SQLAlchemy 2.0.x pour les systèmes de production jusqu’à ce que la version 2.1 atteigne la disponibilité générale
Fixez votre dépendance à une version spécifique (par exemple, sqlalchemy==2.1.0b2) plutôt que d’utiliser des plages de versions

Voir la section Limitations connues pour plus de détails sur le moment d’utilisation des versions pré-lancement.

Logiciels requis

Python 3.10 ou version ultérieure. SQLAlchemy 2.1 a abandonné le support de Python 3.9 et versions antérieures.
Les paquets mssql-python et sqlalchemy (version 2.1.0b2 ou ultérieure).

Les exemples de cet article utilisent la base de données d’exemples AdventureWorksLT . Si vous n’avez pas AdventureWorksLT installé, consultez les bases de données d’exemple AdventureWorks.

Installer la version préliminaire

Comme SQLAlchemy 2.1 est en bêta, pip install sqlalchemy il installe par défaut la dernière version stable 2.0.x. Installez explicitement la préversion :

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

Vérifiez la version installée :

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

URL de connexion

Le dialecte mssql-python utilise mssql+mssqlpython comme schéma d’URL. Le format général est le suivant :

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

Authentification SQL

Pour l’authentification SQL, incluez le nom d’utilisateur et le mot de passe dans l’URL de connexion :

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

Authentification par Microsoft Entra

Pour l’authentification Microsoft Entra, utilisez un nom d’utilisateur vide et le authentication paramètre de requête :

from sqlalchemy import create_engine

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

Note

ActiveDirectoryDefault utilise DefaultAzureCredential, qui tente successivement d’utiliser plusieurs fournisseurs d’informations d’identification. La première connexion peut être lente car le SDK parcourt la chaîne jusqu’à ce qu’il trouve un fournisseur fonctionnel. En production, si vous savez quel type d’identifiant votre environnement utilise, spécifiez-le directement (par exemple, ActiveDirectoryMSI pour l’identité gérée) afin d’éviter la marche en chaîne. Pour plus d’informations, consultez Authentification Microsoft Entra.

Créer des URL par programmation

À utiliser sqlalchemy.engine.URL.create pour éviter le codage manuel des 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)

Définir les modèles ORM

Utilisez la cartographie déclarative de SQLAlchemy pour définir des modèles qui correspondent aux tables SQL 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 utilise IDENTITY l’incrémentation automatique des colonnes. SQLAlchemy mappe automatiquement cela pour les colonnes de clés primaires entières. Les informations explicites Identity() ci-dessus sont optionnelles, sauf si vous devez contrôler les valeurs de début et d’incrément.

Opérations CRUD

Les exemples suivants montrent comment insérer, interroger, mettre à jour et supprimer des lignes en utilisant la session ORM. Chaque exemple réutilise new_id, le ProductID renvoyé lorsque vous insérez une ligne. Pour exécuter les quatre opérations ensemble, voir l’exemple complet.

Créer une session

Créez une session pour exécuter des opérations au sein d’une transaction :

from sqlalchemy.orm import Session

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

Pour les applications qui génèrent de nombreuses sessions, utilisez sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Insérer des lignes

Ajoutez un nouveau produit, validez la session et capturez le ProductID généré dans les exemples suivants :

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

Dans SalesLT.Product, Name et ProductNumber ont tous deux des contraintes uniques. Si vous exécutez cet insert plusieurs fois, changez ces valeurs ou supprimez d’abord la ligne précédente. L’exemple complet supprime la ligne qu’il crée, ce qui permet de s’exécuter à plusieurs reprises.

Lignes de résultats de requête

Récupérer une seule ligne par clé primaire, ou utiliser select() pour des requêtes filtrées :

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

Mettre à jour des lignes

Modifiez un champ dans une ligne existante et validez :

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

Supprimer des lignes

Supprimez une ligne et validez :

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

Exemple complet

Les sections précédentes montraient chaque pièce séparément. Cette section les regroupe en un script autonome que vous pouvez copier, exécuter et relancer.

Créez un fichier nommé crud.py et ajoutez le code suivant. Remplacez les détails create_engine de connexion par les vôtres (voir URL de connexion) :

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

Exécutez le script :

python crud.py

Vous voyez une sortie similaire à la suivante :

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

Le script supprime la ligne qu’il crée, donc il ne rencontre pas les contraintes uniques sur Name et ProductNumber quand vous le relancez. À chaque exécution, une nouvelle ligne est insérée, donc ProductID augmente à chaque fois.

Requêtes principales

SQLAlchemy Core fournit une API d’expression SQL de bas niveau. Vous pouvez utiliser Core avec le même moteur et les mêmes définitions de tables, y compris les classes mappées par ORM.

from sqlalchemy import text

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

Utilisez des constructions de niveau table pour générer du SQL avec sécurité de typage :

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

Lorsque vous sélectionnez des colonnes mappées individuelles dont le nom de la base de données diffère du nom de l’attribut (par exemple, Product.name correspondant à la Name colonne), les lignes principales sont indexées par le nom de la colonne de la base de données. Ajouter .label("name") pour accéder à la valeur au row.name lieu de row.Name.

Regroupement de connexions

SQLAlchemy gère par défaut un pool de connexions. Réglez les paramètres du pool selon votre charge de travail :

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

Paramètre	Description
`pool_size`	Nombre de connexions à garder ouvertes (par défaut : 5).
`max_overflow`	Connexions autorisées au-delà `pool_size` (par défaut : 10).
`pool_timeout`	Quelques secondes pour attendre une connexion avant d’afficher une erreur (par défaut : 30).
`pool_recycle`	Quelques secondes après, une connexion est recyclée (par défaut : -1, désactivée). Définissez cette valeur si votre base de données ferme les connexions inactives.

Utilisation avec des frameworks web

SQLAlchemy est couramment utilisé comme couche de base de données pour Flask et FastAPI. Le dialecte mssql-python fonctionne avec tout framework qui prend en charge SQLAlchemy.

Les extraits suivants illustrent le schéma recommandé d’une session par requête pour chaque framework. Ce sont des fragments illustratifs qui supposent le engine modèle et Product des sections précédentes, pas des applications complètes. Pour des applications complètes et exécutables, voir les articles sur l’intégration FastAPI et l’intégration Flask .

Exemple FastAPI

Utilisez une dépendance de générateur pour fournir une session par requête :

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

Exemple de flasque

Utilisez un gestionnaire de contexte pour lier la session à la requête :

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

Migrations alembiques

Alembic gère les migrations de schémas pour les projets SQLAlchemy , et il fonctionne avec le dialecte mssql-python. La fonction autogenerate d’Alembic compare vos modèles à la base de données en ligne, donc quelques étapes supplémentaires empêchent la possibilité de proposer des modifications aux tables que vous ne gérez pas.

Création d’Alembic

Installez Alembic et initialisez un répertoire de migrations :

pip install alembic
alembic init migrations

Dans alembic.ini, définir l’URL de connexion :

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

Pointez Alembic vers vos modèles

Autogenerate a besoin des métadonnées de vos modèles. Placez les modèles gérés par Alembic dans un module importable, tel que models.py. Parce que l’autogénération propose de supprimer toute colonne qu’un modèle omet, on définit un modèle qui possède entièrement sa table plutôt que de réutiliser le modèle simplifié Product mentionné précédemment dans cet article :

# 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

Par défaut, autogenerate considère comme supprimée chaque table de la base de données qui ne figure pas dans target_metadata et génère drop_table pour cette table. Contre une base de données existante comme AdventureWorksLT, cette action peut faire tomber des dizaines de tables. Ajoutez un include_name filtre pour qu’Alembic ne gère que les tables définies par vos modèles, et relisez toujours le script généré avant de l’appliquer.

Dans migrations/env.py, remplace target_metadata = None par le code suivant. Il importe vos modèles et limite la génération automatique vers les schémas et tables qu’ils définissent :

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

Passe include_name et include_schemas=True à context.configure dans les deux run_migrations_offline et run_migrations_online. Ce include_schemas=True paramètre permet à Alembic de voir des tableaux dans des schémas non par défaut tels que SalesLT:

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

Générer et appliquer une migration

Générez une migration à partir de vos modèles :

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

Alembic détecte la nouvelle table et écrit un script de migration :

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

Le généré upgrade() crée la table, puis downgrade() la supprime :

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

Examinez le script, puis appliquez toutes les migrations en attente :

alembic upgrade head

Différences par rapport au dialecte pyodbc

Si vous migrez depuis mssql+pyodbc, le dialecte mssql-python est similaire car les deux pilotes reposent sur le même cadre ODBC. Principales différences :

Sujet	`mssql+pyodbc`	`mssql+mssqlpython`
Installation du pilote ODBC	Nécessite un pilote ODBC séparé (par exemple, le pilote ODBC 18 pour Microsoft SQL).	Le pilote est inclus. Aucun pilote ODBC séparé n’est nécessaire.
URL de connexion	`mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server`	`mssql+mssqlpython://user:pass@host/db`
`fast_executemany`	Pris en charge par `create_engine(..., fast_executemany=True)`.	Non applicable. Le pilote gère les performances par lots en interne.
Disponibilité	Stable, inclus dans SQLAlchemy depuis la version 1.x.	Préversion (SQLAlchemy 2.1.0b2+).

Limites connues

Le dialecte mssql-python pour SQLAlchemy est en pré-version. Avant d’utiliser en production, comprenez ces implications :

Modifications de l’API : Les signatures de méthodes, les types d’exceptions et le comportement peuvent changer avant la version stable finale. Attachez toujours votre version SQLAlchemy à une version pré-release spécifique (par exemple, sqlalchemy==2.1.0b2) et testez les mises à jour en profondeur.
Test limité : Le dialecte offre moins de tests communautaires que le dialecte stable mssql+pyodbc . Vous pourriez rencontrer des cas particuliers ou des fonctionnalités manquantes.
Lacunes de fonctionnalités : Certaines fonctionnalités avancées de l’ORM ou du Core peuvent ne pas fonctionner. Consultez la documentation dialectale SQLAlchemy MSSQL et testez vos cas d’usage avant de vous engager dans un projet.
Aucune garantie de support : Microsoft et SQLAlchemy offrent un support optimal, mais les problèmes pourraient ne pas être résolus avant la version stable.

Quand utiliser la pré-sortie :

les environnements de développement et de test ;
Projets de preuve de concept
Migration depuis mssql+pyodbc si vous souhaitez éviter la dépendance aux pilotes ODBC externes
Projets où l’on peut répondre aux changements d’API et effectuer des tests de régression

Quand NE PAS utiliser la pré-sortie :

Systèmes de production avec des exigences strictes de stabilité
Applications héritées pluriannuelles où les mises à jour de dépendances sont rares
Charges de travail métier critiques jusqu’à ce que SQLAlchemy 2.1 atteigne une GA stable

Pour connaître l’état le plus récent du dialecte en préversion ainsi que les problèmes connus, consultez le dépôt GitHub mssql-python.

Résolution des problèmes

« Aucun module nommé 'sqlalchemy.dialects.mssql.mssqlpython' »

Cette erreur signifie que la version installée de SQLAlchemy n’inclut pas le dialecte mssql-python. Vérifiez que vous avez la version 2.1.0b2 ou une version ultérieure :

pip install "sqlalchemy>=2.1.0b2"

Échecs de connexion

Si create_engine réussit mais que les requêtes échouent, vérifiez que vos paramètres de connexion fonctionnent directement avec 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()

Si la connexion directe fonctionne mais pas SQLAlchemy, vérifiez si des caractères spéciaux dans votre mot de passe ou le nom du serveur posent des problèmes d’encodage d’URL.

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2026-07-22

Utilisez mssql-python avec SQLAlchemy

Logiciels requis

Installer la version préliminaire

URL de connexion

Authentification SQL

Authentification par Microsoft Entra

Créer des URL par programmation

Définir les modèles ORM

Opérations CRUD

Créer une session

Insérer des lignes

Lignes de résultats de requête

Mettre à jour des lignes

Supprimer des lignes

Exemple complet

Requêtes principales

Regroupement de connexions

Utilisation avec des frameworks web

Exemple FastAPI

Exemple de flasque

Migrations alembiques

Création d’Alembic

Pointez Alembic vers vos modèles

Générer et appliquer une migration

Différences par rapport au dialecte pyodbc

Limites connues

Résolution des problèmes

« Aucun module nommé 'sqlalchemy.dialects.mssql.mssqlpython' »

Échecs de connexion

Contenu connexe

Commentaires

Ressources supplémentaires