Utilize mssql-python com SQLAlchemy

SQLAlchemy é o toolkit de Python ORM e bases de dados mais utilizado. A partir do SQLAlchemy 2.1.0b2, um dialeto incorporado para o driver mssql-python permite usar SQLAlchemy ORM e Core com Microsoft SQL e Base de Dados SQL do Azure.

Pré-requisitos

• Python 3.10 ou posterior. O SQLAlchemy 2.1 deixou de suportar Python 3.9 e anteriores.
• Os pacotes mssql-python e sqlalchemy (2.1.0b2 ou posterior).

Os exemplos deste artigo utilizam a base de dados de exemplo AdventureWorksLT. Se não tiver o AdventureWorksLT instalado, consulte as bases de dados de exemplo do AdventureWorks.

Instalar o pré-lançamento

Como SQLAlchemy 2.1 está em beta, pip install sqlalchemy instala por defeito a versão estável mais recente da série 2.0.x. Instale o pré-lançamento explicitamente:

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

Verifique a versão instalada:

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

URLs de ligação

O dialeto mssql-python utiliza mssql+mssqlpython como esquema de URLs. O formato geral é:

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

Autenticação do SQL

Para autenticação SQL, inclua o nome de utilizador e a palavra-passe na URL da ligação:

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

Autenticação do Microsoft Entra

Para a autenticação Microsoft Entra, utilize um nome de utilizador vazio e o parâmetro de consulta authentication:

from sqlalchemy import create_engine

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

Criar URLs programaticamente

Use sqlalchemy.engine.URL.create para evitar a codificação manual de URLs:

from sqlalchemy.engine import URL

url = URL.create(
    "mssql+mssqlpython",
    username="dbuser",
    password="<password>",
    host="localhost",
    port=1433,
    database="<database>",
)
engine = create_engine(url)

Defina modelos ORM

Utilize o mapeamento declarativo do SQLAlchemy para definir modelos que correspondem a tabelas SQL do 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()
    )

Operações CRUD

Os exemplos seguintes mostram como inserir, consultar, atualizar e eliminar linhas usando a sessão ORM. Cada exemplo reutiliza new_id, o ProductID que é devolvido quando inseres uma linha. Para executar as quatro operações em conjunto, veja o exemplo completo.

Crie uma sessão

Crie uma sessão para executar operações dentro de uma transação:

from sqlalchemy.orm import Session

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

Para aplicações que criam muitas sessões, utilize sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Inserir linhas

Adicione um novo produto, efetue o commit da sessão e capture o ProductID gerado para os exemplos seguintes:

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

Linhas de consulta

Recuperar uma única linha por chave primária, ou usar 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}")

Atualizar linhas

Alterar um campo numa linha existente e confirmar:

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

Excluir linhas

Remove uma linha e compromete-te:

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

Exemplo completo

As secções anteriores mostravam cada peça separadamente. Esta secção junta-os num único script autónomo que pode copiar, executar e executar novamente.

Crie um arquivo chamado crud.py e adicione o código a seguir. Substitua os detalhes da conexão em create_engine pelos seus próprios (consulte URLs de Conexão):

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

Executar o script:

python crud.py

Vê uma saída semelhante à seguinte:

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

O script elimina a linha que cria, por isso não viola as restrições de unicidade em Name e ProductNumber quando o executa novamente. Cada execução insere uma nova linha, por isso, o ProductID aumenta de cada vez.

Consultas principais

SQLAlchemy O Core fornece uma API de expressões SQL de nível inferior. Podes usar o Core com o mesmo motor e definições de tabelas, incluindo classes mapeadas por ORM.

from sqlalchemy import text

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

Use construções ao nível de tabela para geração SQL segura em termos de 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()

Agrupamento de conexões

SQLAlchemy gere um pool de ligações por defeito. Ajuste as definições do pool para a sua carga de trabalho:

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

Utilização com frameworks web

SQLAlchemy é comumente usado como camada de base de dados para Flask e FastAPI. O dialeto mssql-python funciona com qualquer framework que suporte SQLAlchemy.

Os excertos seguintes mostram o padrão recomendado de sessão por pedido para cada um dos frameworks. São fragmentos ilustrativos que assumem o engine modelo e Product das secções anteriores, não aplicações completas. Para aplicações completas e executáveis, consulte os artigos sobre integração FastAPI e integração com Flask .

Exemplo do FastAPI

Use uma dependência de gerador para fornecer uma sessão por pedido:

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

Exemplo de Flask

Use um gestor de contexto para definir o âmbito da sessão ao pedido:

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

Migrações alembicas

A Alembic gere migrações de esquemas para projetos SQLAlchemy e funciona com o dialeto mssql-python. A funcionalidade de autogeração do Alembic compara os seus modelos com a base de dados em tempo real, por isso alguns passos extra impedem que proponha alterações em tabelas que não gere.

Configurar Alembic

Instale o Alembic e inicialize um diretório de migrações:

pip install alembic
alembic init migrations

Em alembic.ini, defina a URL da ligação:

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

Aponte Alembic para os seus modelos

O Autogenerate precisa dos metadados dos seus modelos. Coloque os modelos geridos pelo Alembic num módulo importável, como models.py. Como o autogenerate propõe eliminar qualquer coluna que um modelo omita, defina um modelo que seja totalmente dono da sua tabela em vez de reutilizar o modelo simplificado Product mencionado anteriormente neste artigo:

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

Em migrations/env.py, substitua target_metadata = None pelo seguinte código. Importa os teus modelos e limita a autogeração para os esquemas e tabelas que definem:

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 e include_schemas=True para context.configure em ambos run_migrations_offline e run_migrations_online. A include_schemas=True definição permite que a Alembic veja tabelas em esquemas não padrão, tais como SalesLT:

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

Gerar e aplicar uma migração

Gera uma migração a partir dos teus modelos:

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

O Alembic deteta a nova tabela e escreve um script de migração:

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

A instrução gerada upgrade() cria a tabela e downgrade() elimina-a:

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

Revê o script e depois aplica todas as migrações pendentes:

alembic upgrade head

Diferenças em relação ao dialeto pyodbc

Se estás a migrar de mssql+pyodbc, o dialeto mssql-python é semelhante porque ambos os drivers se baseiam no mesmo framework ODBC. Principais diferenças:

Limitações conhecidas

O dialeto mssql-python para SQLAlchemy está em pré-lançamento. Antes de usar em produção, compreenda estas implicações:

• Alterações na API: Assinaturas de métodos, tipos de exceção e comportamento podem mudar antes da versão estável final. Fixa sempre a tua versão SQLAlchemy a uma build pré-lançamento específica (por exemplo, sqlalchemy==2.1.0b2) e testa as atualizações a fundo.

• Testes Limitados: O dialeto tem menos testes comunitários do que o dialeto estável mssql+pyodbc . Pode encontrar casos limite ou funcionalidades em falta.

• Lacunas de funcionalidades: Algumas funcionalidades avançadas do ORM ou Core podem não funcionar. Consulta a documentação do dialecto SQLAlchemy MSSQL e testa os teus casos de uso antes de te comprometeres com um projeto.

• Sem Garantia de Suporte: Microsoft e SQLAlchemy oferecem suporte de melhor esforço, mas os problemas podem não ser resolvidos antes da versão estável.

Quando Usar o Pré-Lançamento:

• Ambientes de desenvolvimento e teste
• Projetos de prova de conceito
• Migrar de mssql+pyodbc para evitar a dependência de controladores ODBC externos
• Projetos onde pode responder a alterações na API e realizar testes de regressão

Quando NÃO usar o pré-lançamento:

• Sistemas de produção com requisitos rigorosos de estabilidade
• Aplicações legadas de vários anos onde as atualizações de dependências são raras
• Cargas de trabalho empresariais críticas até que SQLAlchemy 2.1 atinja a disponibilidade geral estável

Para o estado mais recente dos dialetos pré-lançamento e problemas conhecidos, consulte o repositório mssql-python do GitHub.

Troubleshooting

Não existe nenhum módulo com o nome 'sqlalchemy.dialects.mssql.mssqlpython'

Este erro significa que a versão instalada do SQLAlchemy não inclui o dialeto mssql-python. Verifique se tem a versão 2.1.0b2 ou posterior:

pip install "sqlalchemy>=2.1.0b2"

Falhas de ligação

Se create_engine tiver sucesso mas as consultas falharem, verifique se os seus parâmetros de ligação funcionam diretamente com 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 a ligação direta funcionar mas o SQLAlchemy não, verifique se há problemas de codificação de URLs em caracteres especiais dentro da sua palavra-passe ou nome do servidor.

Conteúdo relacionado

• Cadeias de ligação MSSQL-Python
• Gestão de Conexões MSSQL-Python
• Integração com FastAPI
• Integração com o Flask
• Documentação do dialeto MSSQL do SQLAlchemy