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.

Importante

El dialecto mssql-python se añadió en SQLAlchemy 2.1.0b2 (publicado el 16 de abril de 2026). SQLAlchemy 2.1 es actualmente una serie de pre-lanzamiento y no se recomienda para uso en producción. Antes de actualizar de SQLAlchemy 2.0, ten en cuenta lo siguiente:

Las APIs podrían cambiar antes de la versión estable final (2.1 GA)
Prueba a fondo tu carga de trabajo antes del despliegue
Utiliza la versión estable de SQLAlchemy 2.0.x para sistemas de producción hasta que la versión 2.1 alcance el estado GA
Fija tu dependencia a una versión específica (por ejemplo, sqlalchemy==2.1.0b2) en lugar de usar rangos de versiones

Consulta la sección de Limitaciones Conocidas para detalles sobre cuándo usar versiones previas.

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

Note

ActiveDirectoryDefault utiliza DefaultAzureCredential, que prueba varios proveedores de credenciales de forma secuencial. La primera conexión puede ser lenta porque el SDK recorre la cadena hasta encontrar un proveedor que funcione. En producción, si sabes qué tipo de credencial utiliza tu entorno, especifícala directamente (por ejemplo, ActiveDirectoryMSI para identidad gestionada) para evitar el recorrido en cadena. Para más información, consulte Autenticación de Microsoft Entra.

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

Tip

Microsoft SQL utiliza IDENTITY el incremento automático de columnas. SQLAlchemy mapea esto automáticamente para las columnas clave primarias enteras. La etiqueta explícita Identity() que se muestra arriba es opcional, a menos que necesites controlar los valores de inicio e incremento.

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

Note

En SalesLT.Product, tanto Name como ProductNumber tienen restricciones únicas. Si ejecutas este insert más de una vez, cambia estos valores o elimina primero la fila anterior. El ejemplo completo elimina la fila que crea, por lo que puede ejecutarse repetidamente.

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

Note

Cuando seleccionas columnas mapeadas individuales cuyo nombre de base de datos difiere del nombre del atributo (por ejemplo, Product.name mapeo a la Name columna), las filas Core se codifican por el nombre de columna de la base de datos. Sumar .label("name") para acceder al valor como row.name en lugar de row.Name.

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

Parámetro	Descripción
`pool_size`	Número de conexiones que mantener abiertas (por defecto: 5).
`max_overflow`	Conexiones permitidas más allá `pool_size` (por defecto: 10).
`pool_timeout`	Segundos para esperar una conexión antes de mostrar un error (por defecto: 30).
`pool_recycle`	Segundos después, se recicla una conexión (por defecto: -1, desactivado). Establece este valor si tu base de datos cierra conexiones inactivas.

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

Caution

Por defecto, autogenerate considera eliminada toda tabla de la base de datos que no esté en target_metadata y emite drop_table para cada una. En una base de datos existente como AdventureWorksLT, esa acción puede eliminar decenas de tablas. Añade un include_name filtro para que Alembic gestione solo las tablas que definen tus modelos, y revisa siempre el script generado antes de aplicarlo.

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:

Tema	`mssql+pyodbc`	`mssql+mssqlpython`
Instalación de controladores ODBC	Requiere un controlador ODBC separado (por ejemplo, el controlador ODBC 18 para Microsoft SQL).	El controlador se incluye. No se necesita un controlador ODBC separado.
URL de conexión	`mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server`	`mssql+mssqlpython://user:pass@host/db`
`fast_executemany`	Soportado por `create_engine(..., fast_executemany=True)`.	No es aplicable. El controlador gestiona internamente el procesamiento por lotes.
Disponibilidad	Estable, incluido en SQLAlchemy desde la versión 1.x.	Versión preliminar (SQLAlchemy 2.1.0b2+).

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.

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2026-07-22

Usa mssql-python con SQLAlchemy

Prerequisites

Instala la versión preliminar

URLs de conexión

Autenticación de SQL

autenticación de Microsoft Entra

Crear URL mediante programación

Definir modelos ORM

Operaciones CRUD

Creación de una sesión

Insertar filas

Filas de consulta

Actualizar filas

Eliminar filas

Ejemplo completo

Consultas principales

Agrupación de conexiones

Uso con frameworks web

Ejemplo de FastAPI

Ejemplo de Flask

Migraciones alembicas

Configurar Alembic

Apunta Alembic a tus modelos

Generar y aplicar una migración

Diferencias con el dialecto pyodbc

Limitaciones conocidas

Troubleshooting

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

Fallos de conexión

Contenido relacionado

Comentarios

Recursos adicionales