Usa mssql-python con SQLAlchemy

SQLAlchemy es el toolkit de Python ORM y bases de datos más utilizado. A partir de SQLAlchemy 2.1.0b2, un dialecto integrado para el controlador mssql-python te permite usar SQLAlchemy ORM y Core con Microsoft SQL y Azure SQL Database.

Prerequisites

• Python 3.10 o posterior. SQLAlchemy 2.1 dejó de dar soporte para Python 3.9 y versiones anteriores.
• Los paquetes mssql-python y sqlalchemy (2.1.0b2 o una versión posterior).

Los ejemplos de este artículo utilizan la base de datos de ejemplo AdventureWorksLT . Si no tienes instalado AdventureWorksLT, consulta las bases de datos de ejemplo de AdventureWorks.

Instala la versión preliminar

Como SQLAlchemy 2.1 está en beta, pip install sqlalchemy instala por defecto la última versión estable de la 2.0.x. Instala explícitamente la versión preliminar:

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

Compruebe la versión instalada:

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

URLs de conexión

El dialecto mssql-python utiliza mssql+mssqlpython como esquema de URL. El formato general es:

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

Autenticación de SQL

Para la autenticación SQL, incluye el nombre de usuario y la contraseña en la URL de conexión:

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

autenticación de Microsoft Entra

Para la autenticación de Microsoft Entra, usa un nombre de usuario vacío y el authentication parámetro de consulta:

from sqlalchemy import create_engine

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

Crear URL mediante programación

Úsalo sqlalchemy.engine.URL.create para evitar la codificación manual de 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)

Definir modelos ORM

Utiliza el mapeo declarativo de SQLAlchemy para definir modelos que se asignen a tablas SQL de 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()
    )

Operaciones CRUD

Los siguientes ejemplos muestran cómo insertar, consultar, actualizar y eliminar filas utilizando la sesión ORM. Cada ejemplo reutiliza new_id, el ProductID devuelto cuando insertas una fila. Para ejecutar las cuatro operaciones juntas, véase el ejemplo completo.

Creación de una sesión

Crea una sesión para ejecutar operaciones dentro de una transacción:

from sqlalchemy.orm import Session

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

Para aplicaciones que crean muchas sesiones, utiliza sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Insertar filas

Añade un nuevo producto, confirma la sesión y captura el ProductID generado para los siguientes ejemplos:

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

Filas de consulta

Recupere una sola fila por clave primaria, o use select() para consultas filtradas:

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

Actualizar filas

Modifica un campo en una fila existente y confirma los cambios:

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

Eliminar filas

Elimina una fila y confirma los cambios:

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

Ejemplo completo

Las secciones anteriores mostraban cada pieza por separado. Esta sección los combina en un solo script autónomo que puedes copiar, ejecutar y ejecutar de nuevo.

Cree un archivo denominado crud.py y agregue el código siguiente. Sustituye los datos create_engine de conexión por los tuyos propios (ver URLs de conexión):

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

Ejecute el script:

python crud.py

Ves una salida similar a la siguiente:

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

El script elimina la fila que crea, así que no infringe las restricciones únicas de Name y ProductNumber cuando lo vuelves a ejecutar. Cada ejecución inserta una nueva fila, por lo que ProductID aumenta cada vez.

Consultas principales

SQLAlchemy Core proporciona una API de expresiones SQL de nivel inferior. Puedes usar Core con el mismo motor y definiciones de tablas, incluyendo clases mapeadas por ORM.

from sqlalchemy import text

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

Utiliza construcciones a nivel de tabla para generar SQL seguro en tipos:

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

Agrupación de conexiones

SQLAlchemy gestiona por defecto un pool de conexiones. Ajusta la configuración del pool según tu carga de trabajo:

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

Uso con frameworks web

SQLAlchemy se utiliza comúnmente como capa de base de datos para Flask y FastAPI. El dialecto mssql-python funciona con cualquier framework que soporte SQLAlchemy.

Los siguientes fragmentos muestran el patrón recomendado de una sesión por solicitud para cada marco de trabajo. Son fragmentos ilustrativos que asumen el engine modelo y Product de las secciones anteriores, no aplicaciones completas. Para aplicaciones completas y ejecutables, véase los artículos sobre integración de FastAPI y Flask .

Ejemplo de FastAPI

Utiliza una dependencia del generador para proporcionar una sesión por solicitud:

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

Ejemplo de Flask

Utiliza un gestor de contexto para asignar el alcance de la sesión a la solicitud:

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

Migraciones alembicas

Alembic gestiona migraciones de esquemas para proyectos SQLAlchemy y funciona con el dialecto mssql-python. La función de autogeneración de Alembic compara tus modelos con la base de datos en vivo, así que unos pasos extra evitan que proponga cambios en tablas que no gestionas.

Configurar Alembic

Instala Alembic e inicializa un directorio de migraciones:

pip install alembic
alembic init migrations

En alembic.ini, establece la URL de conexión:

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

Apunta Alembic a tus modelos

La generación automática necesita los metadatos de tus modelos. Coloca los modelos que gestiona Alembic en un módulo importable, como models.py. Como la generación automática propone eliminar cualquier columna que un modelo omita, define un modelo que represente por completo su tabla en lugar de reutilizar el modelo simplificado Product usado anteriormente en este artículo:

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

En migrations/env.py, sustituye target_metadata = None por el siguiente código. Importa tus modelos y limita la generación automática a los esquemas y tablas que definen:

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

Pase include_name y include_schemas=True a context.configure en ambos run_migrations_offline y run_migrations_online. La include_schemas=True configuración permite a Alembic ver tablas en esquemas no predeterminados como SalesLT:

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

Generar y aplicar una migración

Genera una migración desde tus modelos:

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

Alembic detecta la nueva tabla y escribe un script de migración:

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

El código generado por upgrade() crea la tabla y downgrade() la elimina:

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

Revisa el script y luego aplica todas las migraciones pendientes:

alembic upgrade head

Diferencias con el dialecto pyodbc

Si migras desde mssql+pyodbc, el dialecto mssql-python es similar porque ambos controladores se basan en el mismo marco ODBC. Diferencias clave:

Limitaciones conocidas

El dialecto mssql-python para SQLAlchemy está en versión preliminar. Antes de usarlos en producción, entiende estas implicaciones:

• Cambios en la API: Las firmas de métodos, los tipos de excepción y el comportamiento pueden cambiar antes de la versión estable final. Fija siempre tu versión de SQLAlchemy en una compilación preliminar específica (por ejemplo, sqlalchemy==2.1.0b2) y prueba exhaustivamente las actualizaciones.

• Pruebas limitadas: El dialecto tiene menos pruebas comunitarias que el dialecto estable mssql+pyodbc . Podrías encontrarte con casos límite o funciones que falten.

• Carencias de funcionalidades: Es posible que algunas funcionalidades avanzadas de ORM o Core no funcionen. Consulta la documentación del dialecto SQLAlchemy MSSQL y prueba tus casos de uso antes de comprometerte con un proyecto.

• Sin garantía de soporte: Microsoft y SQLAlchemy ofrecen soporte de mayor esfuerzo, pero los problemas pueden no resolverse antes de la versión estable.

Cuándo usar la versión preliminar:

• Entornos de desarrollo y pruebas
• Proyectos de prueba de concepto
• Migrando desde mssql+pyodbc si quieres evitar la dependencia de los drivers ODBC externos
• Proyectos donde puedes responder a cambios en la API y realizar pruebas de regresión

Cuándo NO usar la versión preliminar:

• Sistemas de producción con estrictos requisitos de estabilidad
• Aplicaciones heredadas de varios años donde las actualizaciones de dependencias son raras
• Cargas de trabajo críticas de negocio hasta que SQLAlchemy 2.1 alcance una GA estable

Para el estado más reciente del dialecto previo al lanzamiento y problemas conocidos, consulta el repositorio mssql-python de GitHub.

Troubleshooting

"No hay módulo llamado 'sqlalchemy.dialects.mssql.mssqlpython'"

Este error significa que la versión instalada de SQLAlchemy no incluye el dialecto mssql-python. Verifica que tienes la versión 2.1.0b2 o posterior:

pip install "sqlalchemy>=2.1.0b2"

Fallos de conexión

Si create_engine tiene éxito pero las consultas fallan, verifica que tus parámetros de conexión funcionen directamente 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()

Si la conexión directa funciona pero SQLAlchemy no, comprueba si hay problemas de codificación de URL en caracteres especiales dentro de tu contraseña o nombre del servidor.

Contenido relacionado

• Cadenas de conexión mssql-python
• Gestión de conexiones MSSQL-Python
• Integración con FastAPI
• Integración con Flask
• Documentación del dialecto MSSQL de SQLAlchemy