Use mssql-python com SQLAlchemy

SQLAlchemy é o conjunto de ferramentas de Python ORM e banco de dados mais amplamente utilizado. A partir do SQLAlchemy 2.1.0b2, um dialeto embutido para o driver mssql-python permite usar SQLAlchemy ORM e Core com Microsoft SQL e Banco 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 neste artigo utilizam o banco de dados de exemplos AdventureWorksLT . Se você não tem o AdventureWorksLT instalado, veja bancos de dados de exemplo do AdventureWorks.

Instale o pré-lançamento

Como o SQLAlchemy 2.1 está em beta, pip install sqlalchemy instala a versão mais recente estável da 2.0.x por padrão. 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 conexão

O dialeto mssql-python usa mssql+mssqlpython como esquema de URL. O formato geral é:

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

Autenticação do SQL

Para autenticação SQL, inclua o nome de usuário e a senha na URL da conexã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 autenticação Microsoft Entra, use um nome de usuário vazio e o authentication parâmetro de consulta:

from sqlalchemy import create_engine

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

Crie URLs de forma programática

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

Use o mapeamento declarativo do SQLAlchemy para definir modelos que mapeiam tabelas do Microsoft SQL Server.

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 a seguir mostram como inserir, consultar, atualizar e excluir linhas usando a sessão ORM. Cada exemplo reutiliza new_id, o ProductID retornado ao inserir uma linha. Para rodar as quatro operações juntas, veja o exemplo completo.

Criar 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, use 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 seguintes exemplos:

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

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

Atualizar linhas

Modifique um campo em uma linha existente e confirme:

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

Excluir linhas

Remova uma linha e faça o commit

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

Exemplo completo

As seções anteriores mostravam cada peça separadamente. Esta seção os combina em um único script autônomo que você pode copiar, rodar e executar novamente.

Crie um arquivo nomeado crud.py e adicione o código a seguir. Substitua os detalhes create_engine da conexão pelos seus próprios (veja 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

Você 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 apaga a linha que cria, então não viola as restrições de unicidade em Name e ProductNumber quando você o executa novamente. Cada execução insere uma nova linha, portanto, o ProductID aumenta a cada vez.

Consultas principais

SQLAlchemy O Core oferece uma API de expressões SQL de nível inferior. Você pode 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 em nível de tabela para geração de SQL segura em 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 gerencia um pool de conexões por padrão. Ajuste as configurações do pool para 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,
)

Uso com frameworks web

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

Os trechos a seguir mostram o padrão recomendado de uma sessão por requisição para cada framework. São fragmentos ilustrativos que assumem o engine modelo e Product das seções anteriores, não aplicativos completos. Para aplicações completas e executáveis, veja os artigos sobre integração FastAPI e integração com Flask .

Exemplo do FastAPI

Use uma dependência geradora para fornecer uma sessão por requisição:

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 gerenciador de contexto para delimitar a sessão à requisição:

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

O Alembic gerencia migrações de esquemas para projetos SQLAlchemy e funciona com o dialeto mssql-python. O recurso autogenerate do Alembic compara seus modelos com o banco de dados ativo, então alguns passos extras impedem que ele proponha alterações em tabelas que você não gerencia.

Configurar o 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 conexão:

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

Aponte Alembic para seus modelos

O Autogenerate precisa dos metadados dos seus modelos. Coloque os modelos gerenciados pela Alembic em um módulo importável, como models.py. Como o autogenerate propõe eliminar qualquer coluna que um modelo omita, defina um modelo que possua totalmente 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. Ele importa seus modelos e limita a geração automática para os esquemas e tabelas que eles 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 configuração permite que a Alembic veja tabelas em esquemas não padrão, como SalesLT:

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

Gerar e aplicar uma migração

Gerar uma migração a partir dos seus modelos:

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

O Alembic detecta 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

O comando gerado upgrade() cria a tabela, e downgrade() a exclui:

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

Revise o script e então aplique todas as migrações pendentes:

alembic upgrade head

Diferenças em relação ao dialeto pyodbc

Se você está migrando de mssql+pyodbc, o dialeto mssql-python é semelhante porque ambos os drivers são baseados 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. Sempre fixe sua versão SQLAlchemy em uma build pré-lançamento específica (por exemplo, sqlalchemy==2.1.0b2) e teste as atualizações cuidadosamente.

• Testes Limitados: O dialeto tem menos testes comunitários do que o dialeto estável mssql+pyodbc . Você pode encontrar casos extremos ou funcionalidades ausentes.

• Lacunas de Recursos: Alguns recursos avançados de ORM ou Core podem não funcionar. Consulte a documentação do dialeto SQLAlchemy MSSQL e teste seus casos de uso antes de se comprometer 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
• Migrando de mssql+pyodbc se você quiser evitar a dependência de drivers ODBC externos
• Projetos onde você pode responder a mudanças 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 atualizações de dependência são raras
• Cargas de trabalho empresariais críticas até que SQLAlchemy 2.1 atinja a disponibilidade geral estável

Para consultar o status mais recente dos dialetos em pré-lançamento e os problemas conhecidos, confira o repositório GitHub do mssql-python.

Solução de problemas

Não há nenhum módulo chamado 'sqlalchemy.dialects.mssql.mssqlpython'

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

pip install "sqlalchemy>=2.1.0b2"

Falhas na conexão

Se create_engine der certo e as consultas falharem, verifique se seus parâmetros de conexã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 conexão direta funcionar, mas SQLAlchemy não, verifique se há problemas de codificação de URL em caracteres especiais dentro da sua senha ou nome do servidor.

Conteúdo relacionado

• Cadeias de conexão do mssql-python
• Gerenciamento de Conexões MSSQL-Python
• Integração com FastAPI
• Integração com Flask
• Documentação do dialeto MSSQL do SQLAlchemy