Como utilizar o SQLAlchemy com o banco de dados SQL Server

Hoje veremos como utilizar a linguagem de programação Python com o banco de dados Microsoft SQL Server através do conector pyodbc.

Microsoft SQL Server

O Microsoft SQL Server é um sistema de gerenciamento de banco de dados relacional (RDBMS) desenvolvido pela Microsoft.

Ele é amplamente utilizado por empresas e organizações para armazenar, consultar e gerenciar grandes volumes de dados.

O SQL Server oferece suporte a uma variedade de recursos, incluindo consultas SQL avançadas, procedimentos armazenados, funções definidas pelo usuário, gatilhos, replicação de dados, integração com outras tecnologias da Microsoft, como o .NET Framework, e ferramentas de administração robustas.

O SQL Server está disponível em várias edições, desde a versão Express gratuita até edições empresariais com recursos avançados de escalabilidade e segurança.

Posts relacionados:

Dependências

macOS

brew install \
unixodbc

Linux

Arch Linux

sudo pacman -S \
unixodbc

Fedora

sudo dnf install \
python3-devel \
unixODBC-devel

Ubuntu

sudo apt install \
unixodbc-dev

SQLAlchemy

O SQLAlchemy é uma biblioteca em Python que fornece uma maneira de interagir com bancos de dados relacionais de forma flexível e poderosa.

Ele permite que você trabalhe com bancos de dados SQL de uma maneira orientada a objetos, mapeando tabelas de banco de dados para classes em Python e consultando essas tabelas usando uma sintaxe semelhante ao SQL.

Em resumo, o SQLAlchemy simplifica o acesso e a manipulação de bancos de dados SQL em aplicativos Python.

O SQLAlchemy pode ser instalado usando o pip, que é o gerenciador de pacotes padrão do Python:

pip install SQLAlchemy

Pyodbc

pyodbc é um módulo Python que fornece uma interface para acessar bancos de dados utilizando ODBC (Open Database Connectivity).

O ODBC é uma API padrão da indústria para acessar bancos de dados, independentemente do sistema operacional em que o banco de dados esteja sendo executado.

O pyodbc permite que os desenvolvedores Python se conectem a uma ampla variedade de bancos de dados, como SQL Server, Oracle, MySQL, PostgreSQL, SQLite e outros, usando o mesmo código, desde que o driver ODBC apropriado esteja instalado.

A sua instalação pode ser feita através do gerenciador de pacotes do Python:

pip install pyodbc

Driver

Além do conector é necessária a instalação do driver do SQL Server.

Cada versão do Microsoft SQL Server pode utilizar uma versão especifica do driver:

SQL Server: Lançado com o SQL Server 2000.
SQL Native Client: Lançado com o SQL Server 2005 (Também conhecido como versão 9.0).
SQL Server Native Client 10.0: Lançado com o SQL Server 2008.
SQL Server Native Client 11.0: Lançado com o SQL Server 2012.
ODBC Driver 11 for SQL Server: Suporta o SQL Server 2005 até 2014.
ODBC Driver 13 for SQL Server: Suporta o SQL Server 2005 até 2016.
ODBC Driver 13.1 for SQL Server: Suporta o SQL Server 2008 Até 2016.
ODBC Driver 17 for SQL Server: Suporta SQL Server 2008 Até 2017.
ODBC Driver 17 for SQL Server: Suporta SQL Server 2008 Até 2017.
ODBC Driver 18 for SQL Server: Suporta SQL Server 2019 ou superior.

No Microsoft Windows é possível realizar a gestão dos drivers através do aplicativo aplicativo Administrador de Fonte de Dados ODBC:

Administrador de Fonte de Dados ODBC no Microsoft Windows.

Contêiner

Para testar o código de exemplo foi utilizando o seguinte arquivo docker-compose.yaml:

services:
  db:
    image: mcr.microsoft.com/mssql/server:2022-latest
    container_name: SQLServer
    restart: on-failure
    ports:
      - '1433:1433'
    environment:
      ACCEPT_EULA: 'Y'
      SA_PASSWORD: 'Docker.123456'
      MSSQL_PID: 'Developer'
      # MSSQL_PID: 'Express'

📝 Contêiner testado com Docker e com o Podman.

Exemplo de CRUD

# -*- coding: utf-8 -*-
'''CRUD - SQLAlchemy - pyodbc - SQL Server (MSSQL).'''

from sqlalchemy import (SmallInteger, String, URL,
                        create_engine, insert, select, delete, update)
from sqlalchemy.orm import DeclarativeBase, Mapped, sessionmaker, mapped_column

url_object = URL.create(
    'mssql+pyodbc',
    username='sa',
    password='Docker.123456',
    host='localhost',
    port=1433,
    database='master',
    query={
        'driver': 'ODBC Driver 18 for SQL Server',
        'TrustServerCertificate': 'yes',
        # 'authentication': 'ActiveDirectoryIntegrated',
    },
)

engine = create_engine(url_object)

Session = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


class TableName(Base):
    __tablename__ = 'table_name'

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column('name', String(32))
    age: Mapped[int] = mapped_column('age', SmallInteger)

    def __repr__(self) -> str:
        return f'TableName(id={self.id}, name={self.name}, age={self.age})'


if __name__ == '__main__':
    # Removing all tables from the database.
    Base.metadata.drop_all(engine)

    # Creating all tables.
    Base.metadata.create_all(engine)

    # Creating a session (add, commit, query, etc.).
    session = Session()

    # Create.
    print('[!] Create [!]')
    session.execute(
        insert(TableName)
        .values(
            name='renato',
            age=35,
        ),
    )

    # Returning the object that will be created.
    result = session.scalar(
        insert(TableName)
        .values(
            name='josé',
            age=29,
        )
        .returning(TableName),
    )
    session.commit()
    print(result)

    # Bulk create.
    session.execute(
        insert(TableName),
        [
            {'name': 'maria', 'age': 25},
            {'name': 'sandy', 'age': 19},
        ],
    )

    result = session.scalars(
        insert(TableName)
        .returning(TableName),
        [
            {'name': 'patrick', 'age': 33},
            {'name': 'gisele', 'age': 21},
        ],
    )
    session.commit()
    print(result.all())

    # Read.
    print('\n[!] Read [!]')
    result = session.scalars(
        select(TableName),
    )
    print(result.all())

    # Get by id.
    result = session.get(TableName, 1)
    print(result)

    # Limit.
    result = session.scalars(
        select(TableName)
        .limit(3),
    )
    print(result.all())

    # Where.
    result = session.scalars(
        select(TableName)
        .where(TableName.age > 30),
    )
    print(result.all())

    result = session.scalar(
        select(TableName)
        .where(TableName.id == 1),
    )
    print(result)

    # Filter.
    result = session.scalars(
        select(TableName)
        .filter_by(name='renato'),
    )
    print(result.all())

    # Update.
    print('\n[!] Update [!]')
    print(session.get(TableName, 1))
    session.execute(
        update(TableName)
        .where(TableName.id == 1)
        .values(name='joão'),
    )
    print(session.get(TableName, 1))

    print(session.get(TableName, 2))
    result = session.scalar(
        select(TableName)
        .where(TableName.id == 2),
    )
    result.name = 'antônio'
    session.commit()
    print(session.get(TableName, 2))

    # Delete.
    print('\n[!] Delete [!]')
    print(session.get(TableName, 1))
    session.execute(
        delete(TableName)
        .where(TableName.id == 1),
    )
    print(session.get(TableName, 1))

    print(session.get(TableName, 2))
    result = session.scalar(
        select(TableName)
        .where(TableName.id == 2),
    )
    session.delete(result)
    session.commit()
    print(session.get(TableName, 2))

    session.close()

Conclusão

Em resumo, um desenvolvedor pode escolher o Microsoft SQL Server devido à sua integração com o ecossistema Microsoft, desempenho e escalabilidade robustos, recursos avançados, segurança sólida e suporte técnico confiável.

Esses aspectos tornam o SQL Server uma escolha atraente para projetos que exigem essas características específicas.