SQLAlchemy ORM

O SQLAlchemy Object Relational Mapper fornece métodos de associação de classes Python definidas pelo usuário com tabelas de banco de dados e instâncias dessas classes (objetos) com linhas em suas tabelas correspondentes, tipos de dados, vínculos e relacionamentos. Ele sincroniza de forma transparente todas as mudanças de estado entre objetos e suas linhas relacionadas, e inclui uma forma de expressação de consultas ao banco de dados como manipulações das classes do Python.

O ORM está construído sobre a SQLAlchemy Expression Language (CORE) mas enfatizando muito mais o modelo definido pelo usuário, garantindo a sincronia entre as duas camadas. Um aplicativo pode ser construído com o uso exclusivo do ORM, embora existam situções em que a Expression Language pode ser usada para fazer interações específicas com o banco de dados.

CREATE TABLE: Para ilustrar a criação de tabelas, inserção de dados, alteração e apagamento de valores listamos aqui o código do python. Outputs de código são precedidos pelo sinal ↳ e os comandos SQL emitidos internamente em quadros iniciados por [SQL].

A classe DeclarativeBase é a base de todas as classes que geram as classes do Python mapeadas em tabelas do banco de dados. Os tipos de cada coluna são informados com anotações com tipos tratados por Mapped[type], onde type é
int (INTEGER), str (VARCHAR), etc. Campos que podem ser nulos são declarados com o modificador Optional[type] (caso contrário o campo é NOT NULL).

A função mapped_column informa tipos e todos os demais atributos da coluna, como a informação de que ela é uma chave estrangeira. A função relationship() estabelece relacionamentos entre classes (portanto entre campos das tabelas). O método de classe __repr__() não é obrigatório mas pode ser útil para debugging. O parâmetro echo=True faz com que o comando SQL subjacente seja exibido no console.

from typing import Optional
from sqlalchemy import create_engine, ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

engine = create_engine('sqlite:///contatos.db' , echo=True)

class Base(DeclarativeBase):
    pass

class Pessoa(Base):
    __tablename__ = "pessoas"

    id: Mapped[int] = mapped_column(primary_key=True)
    nome: Mapped[str] = mapped_column(String(30))
    sobrenome: Mapped[Optional[str]]
  
    enderecos: Mapped[list["Endereco"]] = relationship(back_populates="pessoa",
               cascade="all, delete-orphan")
    def __repr__(self) -> str:
        return f"Pessoa (id = {self.id!r}, nome={self.nome!r}, sobrenome={self.sobrenome!r})"

class Endereco(Base):
    __tablename__ = "enderecos"

    id: Mapped[int] = mapped_column(primary_key=True)
    pessoa_id: Mapped[int] = mapped_column(ForeignKey("pessoas.id"))

    email: Mapped[Optional[str]]
    endereco: Mapped[Optional[str]]
    pessoa: Mapped["Pessoa"] = relationship(back_populates="enderecos")

    def __repr__(self) -> str:
        return f"Endereco: (id = {self.id!r}, email = {self.email!r})"

Base.metadata.create_all(engine)

O comando Base.metadata.create_all(engine) cria um banco de dados e tabelas, se elas não existirem previamente. Os seguintes comandos são gerados.

[SQL]
CREATE TABLE pessoas (
	id INTEGER NOT NULL, 
	nome VARCHAR(30) NOT NULL, 
	sobrenome VARCHAR, 
	PRIMARY KEY (id)
)
CREATE TABLE enderecos (
	id INTEGER NOT NULL, 
	pessoa_id INTEGER NOT NULL, 
	email VARCHAR, 
	endereco VARCHAR, 
	PRIMARY KEY (id), 
	FOREIGN KEY(pessoa_id) REFERENCES pessoas (id)
)

Essa estrutura é denominada Mapeamento Declarativo (Declarative Mapping), responsável pela definição das classes Python e das tabelas, campos e relacionamentos que ficam armazenados em um objeto MetaData (embora esse não seja mencionado explicitamente no código). Temos, como resultado, a criação das tabelas e campos:

pessoas
id
nome
sobrenome

enderecos
id
pessoa_id
email
endereco

INSERT: Para inserirmos valores nas tabelas instanciamos objetos das classes Pessoa e Endereco (que são atribuidos ao campo Pessoa.enderecos). Criamos um objeto session = Session(engine) (dentro de um gerenciador de contexto width) e depois acrescentamos os objetos à sessão com session.add_all([lista_de_objetos]). Nenhuma alteração é gravada no banco de dados até a emissão de session.commit().

galileu = Pessoa(nome="Galileu", sobrenome="Galilei")
paulo = Pessoa(
    nome="Paul",
    sobrenome="Adrian Dirac",
    enderecos=[Endereco(email="pamdirac@hotmail.com")],
)
alberto = Pessoa(
    nome="Albert",
    sobrenome="Einstein",
    enderecos=[Endereco(email="albert@tre.org")],
)
ricardo = Pessoa(
    nome="Richerd",
    sobrenome="Feynman",
    enderecos=[
        Endereco(email="feynman@caltech.edu", endereco="R. Bahia, 2311"),
        Endereco(email="richar@google.com"),
    ],
)

width Session(engine) as session:
    session.add_all([paulo, alberto, ricardo, galileu])
    session.commit()

O nome Richerd foi digitado com erro propositalmente. As consultas são emitidas:

[SQL]
INSERT INTO pessoas (nome, sobrenome) VALUES (?, ?), (?, ?), (?, ?), (?, ?)
    ('Paul', 'Adrian Dirac', 'Albert', 'Einstein', 'Richerd', 'Feynman', 'Galileu', 'Galilei')

INSERT INTO enderecos (pessoa_id, email, endereco) VALUES (?, ?, ?), (?, ?, ?), (?, ?, ?), (?, ?, ?)
(5, 'pamdirac@hotmail.com', None, 6, 'albert@tre.org', None, 7, 'feynman@caltech.edu', 'R. Bahia, 2311', 7, 'richar@google.com', None)

Como resultado temos as tabelas com os seguintes valores:

id	nome	sobrenome
1	Paul	Adrian Dirac
2	Albert	Einstein
3	Richerd	Feynman
4	Galileu	Galilei

id	pessoa_id	email	endereco
1	1	pamdirac@hotmail.com	NULL
2	2	albert@tre.org	NULL
3	3	feynman@caltech.edu	R. Bahia, 2311
4	3	richar@google.com	NULL

SELECT: Consultas podem ser feitas com a classe select. Uma query tem a sintaxe básica query = select(Classe_tabela).where(condicao_na_classe). o resultado é um iterável:

from sqlalchemy import select

session = Session(engine)
query = select(Pessoa).where(Pessoa.nome.in_(["Galileu", "Paul"]))

for p in session.scalars(query):
    print(p)
↳   Pessoa (id = 1, nome='Paul', sobrenome='Adrian Dirac')
    Pessoa (id = 4, nome='Galileu', sobrenome='Galilei')    

A consulta equivalente é:

[SQL]
SELECT pessoas.id, pessoas.nome, pessoas.sobrenome FROM pessoas
    WHERE pessoas.nome IN (?, ?) ('Galileu', 'Paul')

Uma consulta SELECT * pode ser feita diretamente por id:

print(session.get(Pessoa, 4))
↳ Pessoa (id = 4, nome='Galileu', sobrenome='Galilei')
print(session.get(Pessoa, 1).sobrenome)
↳ Adrian Dirac

JOIN: Para realizar consulta com relacionamentos usamos join.

query = (select(Endereco)
    .join(Endereco.pessoa)
    .where(Pessoa.nome == "Richard")
    .where(Endereco.email == "richar@google.com")
)
result = session.scalars(query).one()

print(result)
↳ Endereco: (id = 4, email = 'richar@google.com')

[SQL]
SELECT enderecos.id, enderecos.pessoa_id, enderecos.email, enderecos.endereco 
    FROM enderecos JOIN pessoas ON pessoas.id = enderecos.pessoa_id 
    WHERE pessoas.nome = ? AND enderecos.email = ? ('Richard', 'richar@google.com')

O resultado de print acima decorre da forma como definimos o método __repr__. Qualquer propriedade do objeto pode ser obtida, por exemplo com print(result.id). Em particular result.pessoa é o objeto pessoa associado a esse endereço e print(result.pessoa.nome) imprime o nome “Richard”.

UPDATE: Para alterar um campo de um registro recuperamos o objeto correpondente ao registro e alteramos a propriedade desejada. A alteração só é gravada no BD com session.commit(), quando é emitido e executado o UPDATE.

rick = session.execute(select(Pessoa).filter_by(nome="Richerd")).scalar_one()
print(rick)
↳ Pessoa (id = 3, nome='Richerd', sobrenome='Feynman')

rick.nome = "Richard"
print(rick in session.dirty)
↳ True

# para verificar a alteração (na classe)
rick_nome = session.execute(select(Pessoa.nome).where(Pessoa.id == 3)).scalar_one()
print(rick_nome)
↳ Richard

print(rick in session.dirty)
↳ False

session.commit()

O modificador scalar_one() só pode ser usado quando a consulta retorna apenas uma linha (um objeto). Caso contrário uma exceção é lançada. Após a alteração o objeto fica na coleção Session.dirty até que um commit seja emitido. No caso acima o commit foi implícito, ocorrido quando a query SELECT foi executada.
A consulta resulta em:

[SQL]
SELECT pessoas.id, pessoas.nome, pessoas.sobrenome FROM pessoas 
    WHERE pessoas.nome = ? ('Richerd',)
# depois
UPDATE pessoas SET nome=? WHERE pessoas.id = ? ('Richard', 3)

Uma alteração em um campo exige a recuperação desse objeto seguida da alteração propriamente dita depois a gravação no BD.

query = select(Pessoa).where(Pessoa.id == 3)
p = session.scalars(query).one()
p.sobrenome = "Dawkings"
print(p)
↳ Pessoa (id = 3, nome='Richard', sobrenome='Dawkings')

# para gravar no BD
session.commit()

As consultas são emitidas:

[SQL]
SELECT pessoas.id, pessoas.nome, pessoas.sobrenome  FROM pessoas 
    WHERE pessoas.id = ? (3,)

UPDATE pessoas SET sobrenome=? WHERE pessoas.id = ? ('Dawkings', 3)

DELETE: para uma operação de apagamento de uma linha de tabela recuperamos essa linha (em um objeto) e a apagamos com session.delete(objeto).

p = session.get(Pessoa, 1)
print(p)
↳ Pessoa (id = 1, nome='Paul', sobrenome='Adrian Dirac')

session.delete(p)
session.commit()

Os seguintes comandos SQL são gerados:

[SQL]
SELECT pessoas.id AS pessoas_id, pessoas.nome AS pessoas_nome, pessoas.sobrenome AS pessoas_sobrenome 
   FROM pessoas WHERE pessoas.id = ? (1,)

DELETE FROM enderecos WHERE enderecos.id = ? (1,)
DELETE FROM pessoas WHERE pessoas.id = ? (1,)

Devido aos vínculos estabelecidos na definição da tabela (e, portanto, também da classe) enderecos, relationship(back_populates="pessoa", cascade="all, delete-orphan") ao ser apagada a linha da pessoa de id = 1 as linhas vinculadas da tabela enderecos também são apagadas.