Oharra
Baimena behar duzu orria atzitzeko. Direktorioetan saioa has dezakezu edo haiek alda ditzakezu.
Baimena behar duzu orria atzitzeko. Direktorioak alda ditzakezu.
SQLAlchemy es el toolkit de Python ORM y bases de datos más utilizado.
SQLAlchemy 2.1 incluye un dialecto integrado para el mssql-python controlador que puedes usar para trabajar con SQLAlchemy ORM y Core con Microsoft SQL y Azure SQL Database.
Prerequisites
- Python 3.11 o posterior. SQLAlchemy 2.1 dejó de soportar Python 3.10 y versiones anteriores.
- Los paquetes
mssql-pythonysqlalchemy(versión 2.1 o 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.
Instalar SQLAlchemy y mssql-python
Utiliza la dependencia opcional de mssql-pythonSQLAlchemy para instalar ambos paquetes:
pip install "sqlalchemy[mssql-python]>=2.1"
SQLAlchemy define el mssql-python extra con mssql-python>=1.9.0. También se soporta el formulario existente de paquetes separados:
pip install "sqlalchemy>=2.1" mssql-python
Compruebe la versión instalada:
import sqlalchemy
print(sqlalchemy.__version__) # Should show 2.1.0 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 al insertar 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
Eliminar una fila y confirmar:
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,
)
| Parameter | Description |
|---|---|
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:
| Topic | mssql+pyodbc |
mssql+mssqlpython |
|---|---|---|
| Instalación de controladores ODBC | Requiere un controlador ODBC separado (por ejemplo, el controlador ODBC 18 para SQL Server). | Instalado automáticamente como dependencia de paquete. No hace falta instalación separada de drivers ODBC. |
| 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. | Estable, incluido en SQLAlchemy 2.1 o posterior. |
Consideraciones de actualización
SQLAlchemy 2.1 incluye cambios de comportamiento que podrían afectar a la actualización de aplicaciones respecto a versiones anteriores:
- SQLAlchemy 2.1 requiere Python 3.11 o posterior.
- Actualizar aplicaciones SQLAlchemy 1.x a SQLAlchemy 2.0 antes de pasar a la 2.1.
- Prueba las aplicaciones existentes de SQLAlchemy 2.0 frente a los cambios de comportamiento descritos en ¿Qué hay de nuevo en SQLAlchemy 2.1?
Los ejemplos de este artículo utilizan la API síncrona de SQLAlchemy. Las aplicaciones que utilizan la compatibilidad con asyncio de SQLAlchemy deben instalar el extra asyncio porque SQLAlchemy 2.1 ya no instala greenlet por defecto.
Solución de problemas
"No hay módulo llamado 'sqlalchemy.dialects.mssql.mssqlpython'"
Este error significa que la versión instalada de SQLAlchemy no incluye el mssql-python dialecto. Instala SQLAlchemy 2.1 o una versión posterior junto con su dependencia mssql-python:
pip install --upgrade "sqlalchemy[mssql-python]>=2.1"
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.