ORM

ORM e Relacionamentos

Posted on by Guilherme Silva

ORM Manipulação de Objetos

Vimos na seção sobre a definição de tabelas com o ORM como definir classes do Python que podem ser correlacionadas com entidades do SQL por meio do SQLAlchemy. Já fizemos uso, sem explorar muito o assunto, do método relationship() que insere no esquema de uma sessão os relacionamentos entre propriedades dos objetos que serão espelhadas nas tabelas envolvidas.

Recordando, criamos uma classe vazia herdando de DeclarativeBase que será a superclasse para os modelos de tabelas. No nosso exemplo criamos os objetos aluno e endereco

class Base(DeclarativeBase):
    pass

class Aluno(Base):
    __tablename__ = "aluno"
    ...
    enderecos: Mapped[List["Endereco"]] = relationship(back_populates="aluno")
    
class Endereco(Base):
    __tablename__ = "endereco"
    ...
    aluno: Mapped[Aluno] = relationship(back_populates="enderecos")


Vemos que a classe Aluno tem o atributo Aluno.enderecos e a classe Endereco tem o atributo Endereco.aluno, que estão em relacionamento. Vimos também que Mapped informa o tipo do campo. Objetos da classe Endereco se referem a uma tabela com a campo aluno que é uma chave estrangeira (ForeignKeyConstraint) ligada ao campo aluno.enderecos. O método relationship() pode determinar sem ambiguidade que existe um relacionamento de um para muitos: um aluno.enderecos (uma linha de aluno) pode estar ligada a várias linhas na tabela de endereco.

Relacionamentos um-para-muitos correspondem, é claro, a um relacionamento muitos-para-um na direção oposta. Portanto o parâmetro relacionship.back_populates, em ambas as classes, define que esses campos estão em relação complementar entre si.

Persistência: Um objeto dentro de uma sessão pode ter diversos estados, no que se refere à persistência. Um objeto persistente possui uma identidade em relação ao banco de dados, ou seja, possui uma identidade (uma pk ou chave primária) igual àquela da linha que ele modela. Ao ser criado, antes de ser comitado, um objeto está no estado pendente. Ele se torna persistente com um commit, após ser aplicado no BD. Igualmente, um objeto que foi carregado do BD é persistente. Objetos removidos da sessão são denominados destacados (detached ).

Persistência de relacionamentos: Definidos os relacionamentos eles devem ser gravados na tabela e, quando as tabelas já estão definidas, carregados de volta para as classes do ORM. Suponha que inicializamos um objeto aluno com as seguintes propriedades:

aluno1 = Aluno(matricula="976567-123", nome="Mauro", sobrenome="Olivares")
aluno1.enderecos
↳ []
# uma lista vazia


O campo retornado, inicialmente vazio, é uma versão de uma lista no SQLAlchemy (uma Mapped[List]) que pode rastrear e responder às alterações efetuadas sobre o objeto. Ela é inserido automaticamente quando tentamos acessar o atributo, mesmo que não o tenhamos definido na criação do objeto. Isso é semelhante à inserção de ids que não são informados na incialização. Esse comportamento é diferente daqueles das classes usuais do Python que geram uma exeção AttributeError se a propriedade não for definida na inicialização. O objeto aluno1 é transitório e a lista em aluno1.enderecos não sofreu nenhuma alteração.

Para inserir um elemento nessa coleção criamos um endereço e usamos o método list.append(objeto_endereco).

end1 = Endereco(email="olivares@gmail.com")
aluno1.enderecos.append(end1)

# um endereço é anexado ao objeto aluno1
aluno1.enderecos
↳ [Endereco(id=None, email='olivares@gmail.com')]

# o objeto end1 é sincronizado (veja descrição abaixo)
end1.aluno
↳ Aluno(id=None, nome='Mauro', sobrenome='Olivares')

A operação de inserir um Endereco ao objeto Aluno, além de atualizar o próprio campo aluno1.enderecos também realiza a sincronização automática de Endereco.aluno, inserindo uma referência ao aluno dono desse endereço de email. Essa sincronização é o resultado do parâmetro relationship.back_populates entre os objetos relacionados.

Essa sincronização funciona também na outra direção: se criamos outro objeto Endereco com atributo Endereco.aluno referenciando o aluno1 esse novo endereço fará parte da coleção Aluno.enderecos, para o aluno em questão.

# criamos novo endereco, já associado ao aluno1
end2 = Endereco(email="olivar@aol.com", aluno=aluno1)
# o novo endereco se torna parte da coleção
aluno1.enderecos
↳ [Endereco(id=None, email='olivares@gmail.com'), Endereco(id=None, email='olivar@aol.com')]

Esses novos elementos precisam ser inseridos na sessão, o que pode ser feito com o método session.add(). Com a inserção de aluno1 os dois endereços ficam também inseridos.

session.add(aluno1)
# com esse procedimento temos
aluno1 in session
↳ True
end1 in session
↳ True
end2 in session
↳ True

Essas são as chamadas operações de save e update em cascata. Agora os 3 objetos envolvidos estão em estado pendente: nenhum deles tem um id designado, por enquanto. Além disso os objetos end1 e end2 possuem o atributo aluno_id que é a referência à coluna com um ForeignKeyConstraint ligada à aluno.id. Esse atributo também não foi ainda atribuído a uma linha real do banco de dados, portanto aluno.id = None.

print(aluno1.id)
↳ None
print(end1.aluno_id)
↳ None


Quando comitamos as transações os passos ocorrem na ordem correta, gerenciados pelo SQLAlchemy, para gerar as ids e propagar essa informação para os campos relacionados.

session.commit()
[SQL]
INSERT INTO aluno (nome, sobrenome) VALUES (?, ?) ('Mauro', 'Olivares')
INSERT INTO endereco (email, aluno_id) VALUES (?, ?), (?, ?) RETURNING id
('olivares@gmail.com', 6, 'olivar@aol.com', 6)
COMMIT

No último insert estamos supondo que o id de aluno recém inserido seja 6.

Carregando Relacionamentos: Após a emissão de Session.commit() é emitida automaticamemnte um Session.commit.expire_on_commit que faz com que todos os objetos da sessão fiquem expirados. No próximo acesso de um atributo desses objetos um SELECT é emitido para a linha, permitindo a visualização da chave primária recém-gerada.

aluno1.id
↳ 6
[SQL]
SELECT aluno.id AS aluno_id, aluno.nome AS aluno_nome, aluno.sobrenome AS aluno_sobrenome
FROM aluno WHERE aluno.id = ? (6,)

Podemos também acessar a coleção persistente aluno.enderecos de aluno1, que consiste em um conjunto adicional de linhas da tabela de endereços. Quando acessamos essa coleção ocorre uma lazy load † (uma carga lenta) emitida para recuperar os objetos:

aluno1.enderecos
↳ [Endereco(id=4, email='olivares@gmail.com'), Endereco(id=5, email='olivar@aol.com')]
[SQL]
SELECT endereco.id AS endereco_id, endereco.email AS endereco_email,
endereco.aluno_id AS endereco_aluno_id
FROM endereco WHERE endereco.aluno_id = ? (6,)

† lazy load, eager load: No ORM uma “carga lenta”, ou lazy load, se refere a um atributo que não contém seu valor imediatamente lido no banco de dados. Geralmente isso ocorre quando o objeto é carregado pela primeira vez. O atributo recebe uma referência na memória que permite que ele leia o valor no banco de dados quando for usado pela primeira vez. Esse padrão busca reduzir o tempo gasto nas buscas de objetos que não precisam ser imediatamente exibidos. Carregamentos que ocorrem no momento da chamada são denominados “carregamentos imediatos ou rápidos”, eager load.

As coleções e atributos relacionados no SQLAlchemy ORM são persistentes na memória. Depois que o valor é atribuído não há mais necessidade de emitir consultas SQL até que a coleção ou atributo expire. Podemos acessar, adicionar ou remover itens em aluno1.enderecos sem que novas consultas SQL sejam executadas.

Esse carregamento lento pode se tornar pesado na memória se não forem tomadas medidas para otimizá-lo. Existe otimização para evitar trabalho redundante: a coleção aluno1.enderecos foi atualizada no mapa de identidade onde todas as referências apontam para as mesmas instâncias Endereco já criadas. Portanto, todos esses objetos já estão carregados.

Consultas com Relacionamentos: O SQLAlchemy admite diversos recursos para a construção consultas SQL que envolvem classes mapeadas coom relacionamentos. Os métodos Select.join() e Select.join_from() são usados para compor cláusulas JOIN nas consultas. Esses métodos inferem a cláusula ON com base na presença de um único e inequívoco objeto ForeignKeyConstraint quando constroem consultas com junções (JOIN), vinculando as tabelas à partir da estrutura dos metadados da sessão. Se desejado, também é possível fornecer explicitamente uma expressão SQL especificando a cláusula ON.

Outro mecanismo também está disponível para estabelecer junções quando usamos entidades ORM, usando os objetos gerados por relationship(), que foram configurados no mapeamento das classes. O atributo da classe que está em relacionamento, definido em relationship(), pode ser passado como argumento para Select.join(), para indicar tanto o lado direito da junção quanto o campo na cláusula ON.

# consulta (1)
print(select(Endereco.email).select_from(Aluno).join(Aluno.enderecos))
# consulta (2)
print(select(Endereco.email).join_from(Aluno, Endereco))
# ambas as consultas geram:
[SQL]
SELECT endereco.email FROM aluno JOIN endereco ON aluno.id = endereco.aluno_id

Consultas com Select.join() ou Select.join_from() não usam o relacionamento estabelecido no mapeamento para inferir a cláusula ON, exceto se isso for explicitamente especificado. Isso significa que, quando fazemos união de Aluno para Endereco sem incluir uma cláusula ON, uma consulta correta é emitida por causa da ForeignKeyConstraint entre os objetos mapeados e não devido à existência de um relationship().


Vale lembrar que Aluno, Endereco (com maiúsculas) se referem às classes do ORM enquanto aluno, endereco são os nomes das tabelas no BD.Consulte o manual do SQLAlchemy: ORM Query Guide, Select Join, Select Join On Clause.

Relacionamentos e WHERE

Existem algumas formas de gerar consultas e filtros com relationship(), tipicamente aplicados com WHERE (no SQL) e Select.where() (no SQLAlchemy).

EXISTS: has() e any(): Vimos na seção Agrupamentos e Subqueries: EXISTS como funciona EXISTS e sua contraparte no SQLAlchemy. O método exists() é usado para gerar a cláusula EXISTS do SQL que é aplicada sobre um conjunto de resultados obtidos com uma subconsulta escalar. A classe construída por relationship() tem métodos auxiliares responsáveis pela geração de algumas formas comuns de uso de EXISTS em consultas sobre colunas ligadas por relacionamentos.

Em um relacionamento um-para-muitos, como é o caso de Aluno.enderecos que se liga a uma coleção de Endereco.aluno, podemos gerar um EXISTS usando PropComparator.any(). Este método aceita um critério WHERE opcional para filtrar as linhas retornadas pela subconsulta.

query = select(Aluno.sobrenome)
             .where(Aluno.enderecos.any(Enderecos.email == "olivares@gmail.com"))
session.execute(query).all()
# é retornado
↳ ['Olivares',)]
[SQL]
SELECT aluno.sobrenome FROM aluno
WHERE EXISTS (SELECT 1 FROM endereco WHERE aluno.id =
              endereco.aluno_id AND aluno.email = ?) ('olivares@gmail.com',)

A subconsulta retorna 1 para cada linha que satisfaz
aluno.id = endereco.aluno_id AND aluno.email = 'olivares@gmail.com' .
Se existir algum valor EXISTS retorna TRUE e o sobrenome é retornado pela consulta externa.

O uso de EXISTS é, em geral, mais eficiente para pesquisas negativas, quando se faz uma busca por elementos que não estão presentes nas linhas. Para isso basta negar um resultado, como ~Aluno.endereco.any(), para selecionar Alunos que não possuem linhas associadadas na tabela endereco.

query = select(Aluno.nome).where(~Aluno.enderecos.any())
session.execute(query).all()

[SQL]
SELECT aluno.nome FROM aluno WHERE NOT (EXISTS 
      (SELECT 1 FROM aluno WHERE aluno.id = endereco.aluno_id)
)

A consulta retorna os nomes dos alunos sem um endereço cadastrado.

O método PropComparator.has() age quase da mesma forma que PropComparator.any(), com a diferença de ser usado em relacionamentos muitos-para-um. Esse seria o caso se quisermos encontrar todos os endereços associados com um aluno determinado.

query = select(Endereco.email).where(Endereco.aluno.has(Aluno.nome == "Mauro"))
session.execute(query).all()

[SQL]
SELECT endereco.email FROM endereco WHERE EXISTS
   (SELECT 1 FROM aluno WHERE aluno.id = endereco.aluno_id AND aluno.nome = ?) ('Mauro',)

↳ [('olivares@gmail.com',), ('olivar@aol.com',)]

As consultas 1-4 abaixo exibem outras propriedades: (1) Uma instância de um objeto pode ser comparada a um relacionamento muitos-para-um para selecionar linhas onde a chave estrangeira no destino corresponde à chave primária do objeto dado. (2) O operador not equals (!=) também pode ser usado. (3) Aluno.enderecos.contains(obj_endereco) testa se o objeto carregado é um dos endereços na coleção. (4) with_parent(obj_aluno, Aluno.enderecos) testa se obj_aluno tem Aluno.enderecos como classe pai.

# (1) 
obj_endereco = session.get(Endereco, 1)
obj_aluno = session.get(aluno, 1)
print(select(Endereco).where(Endereco.aluno == obj_aluno))
[SQL]
SELECT endereco.id, endereco.aluno_id, endereco.email 
FROM endereco WHERE :param_1 = endereco.aluno_id

# (2) 
print(select(Endereco).where(Endereco.aluno != obj_aluno))
[SQL]
SELECT endereco.id, endereco.aluno_id, endereco.email FROM endereco
WHERE endereco.aluno_id != :aluno_id_1 OR endereco.aluno_id IS NULL

# (3) 
print(select(Aluno).where(Aluno.enderecos.contains(obj_endereco)))
[SQL]
SELECT aluno.id, aluno.matricula, aluno.nome, aluno.sobrenome
FROM aluno WHERE aluno.id = :param_1

# (4) 
from sqlalchemy.orm import with_parent
print(select(Endereco).where(with_parent(obj_aluno, Aluno.enderecos)))
[SQL]
SELECT endereco.id, endereco.aluno_id, endereco.email
FROM endereco WHERE :param_1 = endereco.aluno_id

Lembramos aqui que Endereco.aluno está em relacionamento (muitos-para-1) com Aluno.enderecos.

Estratégias de de carregamento

Vimos que, quando acessamos atributos de objetos mapeados que usam relacionamentos, um carregamento lento (ou lazy load) será realizado quando a coleção ainda não estiver preenchida. O carregamento lento é um padrão importante no ORM, embora controverso. Quando temos muitos objetos ORM na memória que fazem referência a muitos atributos não carregados, a manipulação desses objetos pode gerar novas consultas em cascata, causando acúmulo (no que consiste no problema denominado “N mais um”). Para piorar o estado de coisas essas novas consultas são emitidas implicitamente. Elas podem causar erros quando são produzidas após o fechamento das transações com o BD, ou quando se usa gerenciadores de conexões assíncronas, como asyncio.

Apesar disso o carregamento lento é útil, principalmente quando bem ajustado com mecanismos de sincronização. Por isso o SQLAlchemy ORM inclui muitos recursos para controlar e otimizar o comportamento de carregamentos. A etapa principal no uso de carregamento lento consiste em testar o aplicativo ativando a exibição de saídas de consultas para análise do SQL emitido. A presença de muitas instruções SELECT redundantes, que poderiam ser agrupadas com mais eficiência, ou a ocorrência de carregamentos inadequados para objetos que já estão destacados (detached ) da sessão, são indicadores de que se deve usar estratégias de carregamento.

Essas estratégias são representadas por objetos que podem ser associados a uma instrução SELECT através do método Select.options(). A estratégia abaixo permite o acesso aos objetos já carregados de Aluno.enderecos.

enderecos_carregados = session.execute(select(Aluno)
                            .options(selectinload(Aluno.enderecos))).scalars()
for obj_aluno in enderecos_carregados:
    obj_aluno.enderecos

Também é possível tornar o carregamento lento a forma default em relationship(), usando a opção relationship.lazy.

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import relationship

class Aluno(Base):
    __tablename__ = "aluno"
    ...
    enderecos: Mapped[List["Endereco"]] = relationship(back_populates="aluno", lazy="selectin")

Carregamento Selectin: Uma opção de carregamento muito útil é a selectinload() que resolve o problema frequente “N mais um”, citado acima. A opção selectinload() faz com que uma coleção completa de objetos relacionados seja carregada antecipadamente em uma única consulta. Isso é obtido com consultas SELECT aplicadas apenas sobre uma tabela, sem inserir JOINs ou subconsultas, seguida de consultas para os objetos relacionados que ainda não foram carregados.

No exemplo abaixo selectinload() é usado para carregar todos os objetos Alunos e os objetos Enderecos associados. Quando invocamos Session.execute() uma vez, passando um select(), o BD dados é acessado com duas instruções SELECT, sendo a segunda usada para carregar objetos Enderecos associados.

from sqlalchemy.orm import selectinload
query = select(Aluno).options(selectinload(Aluno.enderecos)).order_by(Aluno.id)
for row in session.execute(query):
    print(f"Aluno: {row.Aluno.nome} {row.Aluno.sobrenome}")
    for a in row.Aluno.enderecos:
        print(f"{a.email}")
[SQL]
SELECT aluno.id, aluno.nome, aluno.sobrenome
FROM aluno ORDER BY aluno.id

SELECT endereco.aluno_id AS endereco_aluno_id, endereco.id AS endereco_id, 
endereco.email AS endereco_email FROM endereco
WHERE endereco.aluno_id IN (?, ?, ?, ?, ?, ?) (1, 2, 3, 4, 5, 6)

A consulta retorna nome e sobrenome dos 6 primeiros alunos e seus respectivos emails.

Carregamento com JOIN: joinedload() é usado como estratégia de carregamento imediato (eager load ) que inclui a possibilidade de JOINs em uma instrução SELECT. Esse JOIN pode ser uma junção externa ou interna. Essa é a estratégia adequada para carregar objetos em relacionamentos muitos-para-um pois isso exige apenas o carregamento de colunas adicionais a uma linha da entidade primária. Ele também aceita a opção joinload.innerjoin para que a junção seja considerada interna (e não externa). No exemplo abaixo sabemos que todos os objetos Enderecos estão associados a algum Aluno.

from sqlalchemy.orm import joinedload
query = (
    select(Endereco)
    .options(joinedload(Endereco.aluno, innerjoin=True))
    .order_by(Endereco.id)
)
for row in session.execute(query):
    print(f"Aluno: {row.Endereco.aluno.nome}: email: {row.Endereco.email} ")
    
SELECT endereco.id, endereco.email, endereco.aluno_id,
       aluno_1.id AS id_1, aluno_1.nome, aluno_1.sobrenome
FROM endereco JOIN aluno AS aluno_1 ON aluno_1.id = endereco.aluno_id
       ORDER BY endereco.id

A consulta retorna os nomes e emails de alunos, ordenados pelo id do endereço. Lembrando que Endereco está em relação com aluno vemos que Endereco.aluno.nome fica carregado com o nome desse aluno.

joinload() também funciona para coleções, em relacionamentos um-para-muitos. Esse uso, no entanto, pode multiplicar as linhas linhas retornadas de maneira recursiva, o que exige cuidado nessa opção, e consideração do uso de selectinload().

Importante: os critérios WHERE e ORDER BY, usados para modificar a instrução Select, não agem sobre a tabela afetada por joinload(). Como mostra a consulta SQL acima um aliás é atribuído à tabela aluno para que ela não seja alvo desses filtros.

Vemos assim que joinload() recebe como argumento o campo que deve ser carregado de forma imediata. Nos exemplos abaixo os objetos ORM (com letra maiúscula) refletem tabelas Cliente, com campo (uma coleção) Cliente.pedidos; Pedidos com coleção Pedidos.itens, cada item com a descrição Item.descricao.

# joined-load um campo "pedidos" no objeto ORM Cliente
query(Cliente).options(joinedload(Cliente.pedidos))

# joined-load Pedidos.itens, depois Item.descricao (se Pedidos.itens é uma coleção de objetos Item)
query(Pedidos).options(
    joinedload(Pedidos.itens).joinedload(Item.descricao))

# a mesma consulta, com lazy load
query(Pedidos).options(
    lazyload(Pedidos.items).joinedload(Item.descricao))


Junções explícitas com carregamentos rápidos, contains_eager: Suponha que queremos carregar as linhas de endereço associadas à tabela aluno usando um método como Select.join() para aplicar um JOIN. Esse JOIN para ser aproveitado para uma carga rápida do conteúdo de Endereco.aluno em cada campo endereco retornado. Podemos realizar um carregamento rápido como JOIN, executando esse JOIN manualmente. Isso pode ser obtido com contains_eager(), uma opção semelhante a joinload() que libera o desenvolvedor para configurar o JOIN. Colunas adicionais na cláusula COLUMNS devem ser carregadas em atributos relacionados em cada objeto retornado. Por exemplo:

from sqlalchemy.orm import contains_eager
query = (
    select(Endereco)
    .join(Endereco.aluno)
    .where(Aluno.nome == "Marcos")
    .options(contains_eager(Endereco.aluno))
    .order_by(Endereco.id)
)
for row in session.execute(query):
    print(f"{row.Endereco.aluno.nome}, email: {row.Endereco.email} ")
[SQL]
SELECT aluno.id, aluno.nome, aluno.sobrenome, endereco.id AS id_1, endereco.email, endereco.aluno_id
FROM endereco JOIN aluno ON aluno.id = endereco.aluno_id
WHERE aluno.nome = ? ORDER BY endereco.id ('Marcos',)

Filtramos, na consulta acima, as linhas por aluno.nome e carregamos linhas de aluno no atributo Endereco.aluno. Se tivéssemos aplicado joinedload() seriam geradas partes desnecessárias na consulta SQL, como exibido abaixo.

query = (
    select(Endereco)
    .join(Endereco.aluno)
    .where(Aluno.nome == "Marcos")
    .options(joinedload(Endereco.aluno))
    .order_by(Endereco.id)
)
print(query)
[SQL]
SELECT endereco.id, endereco.email, endereco.aluno_id,
aluno_1.id AS id_1, aluno_1.nome, aluno_1.sobrenome
FROM endereco JOIN aluno ON aluno.id = endereco.aluno_id
LEFT OUTER JOIN aluno AS aluno_1 ON aluno_1.id = endereco.aluno_id
WHERE aluno.nome = :nome_1 ORDER BY endereco.id

Esse exemplo produz a geração desnecessária de clásulas JOIN e LEFT OUTER JOIN junto com SELECT.

Raiseload é outra estratégia de carregamento. Ela é usada para impedir o surgimento do problema N-mais-um, transformando cargas lazy em um lançamento de erro. Usamos a opção raiseload.sql_only para bloquear cargas lentas feitas por consultas SQL ou bloquear todos os carregamentos, incluindo aqueles que apenas precisam consultar a sessão atual. Uma das formas consiste em usar raiseload() para configurar o relacionamento estabelecido em relationship(), ajustando o valor relationship.lazy = "raise_on_sql". Com isso nenhum acesso aos dados tentará emitir uma consulta SQL. Isso pode ser feito na definição dos objetos ORM que refletem as tabelas.

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import relationship

class Aluno(Base):
    __tablename__ = "aluno"
    id: Mapped[int] = mapped_column(primary_key=True)
    enderecos: Mapped[List["Endereco"]] = relationship(back_populates="aluno", lazy="raise_on_sql")

class Endereco(Base):
    __tablename__ = "endereco"
    id: Mapped[int] = mapped_column(primary_key=True)
    aluno_id: Mapped[int] = mapped_column(ForeignKey("aluno.id"))
    aluno: Mapped["Aluno"] = relationship(back_populates="enderecos", lazy="raise_on_sql")

Esse tipo de definição no relacionamento impede a realização de “lazy loads” e obriga a definição de uma estratégia de carregamento para consultas nesses campos.

u1 = session.execute(select(Aluno)).scalars().first()
[SQL]
SELECT aluno.id FROM aluno
# ao tentar acessar a propriedade relacionada
u1.enderecos
# um erro é lançado
sqlalchemy.exc.InvalidRequestError: 'Aluno.enderecos' is not available due to lazy='raise_on_sql'

Essa exceção indica que a coleção devaria ter sido carregada antes do uso.

u1 = (
    session.execute(select(User).options(selectinload(User.addresses)))
    .scalars()
    .first()
)
[SQL]
SELECT aluno.id FROM aluno
[...]
SELECT endereco.aluno_id AS endereco_aluno_id, endereco.id AS endereco_id
FROM endereco WHERE endereco.aluno_id IN (?, ?, ?, ?, ?, ?) (1, 2, 3, 4, 5, 6)

O opção lazy="raise_on_sql" também tenta o carregamneto correto em relacionamentos muitos-para-um. Se o atributo Endereco.aluno não estiver preenchido mas o objeto Aluno já está carregado no sessão atual, então a estratégia raiseload não lança erros.

Bibliografia

Esse texto é baseado primariamente na documentação do SQLAlchemy, disponível em SQLAlchemy 2, Documentation. Outras referências no artigo Python e SQL: SQLAlchemy.

SQLAlchemy: ORM

Posted on by Guilherme Silva

ORM

Após termos visto do uso consultas de inserção, alteração e apagamento com o SQLAlchemy Core podemos considerar o mesmo conjunto de operações com o ORM.

Usando ORM

Na abordagem ORM (Object Relational Mapper) do SQLAlchemy o objeto Section é a base da interação entre o código Python e os bancos de dados. Ele é usado de forma muito semelhante ao objeto Connection, usado no CORE que, internamente, é acionado pelas sessões do ORM para produzir consultas SQL.

Para ilustrar o processo básico vamos usar a construção dos padrões semelhante à usada com Connection, usando um gerenciador de contexto, embora Section admita alguns padrões de criação diferentes.

from sqlalchemy.orm import Session
query = text("SELECT campo_1, campo_2 FROM tabela WHERE campo_1 > :c ORDER BY campo_1, campo_2")
with Session(engine) as session:
    result = session.execute(query, {"c": 23})
    for row in result:
        print(f"campo_1: {row.campo_1}  campo_2: {row.campo_2}")
[SQL]
SELECT campo_1, campo_2 FROM tabela WHERE campo_1 > ? ORDER BY campo_1, campo_2
[...] (23,)
# o resultado contém uma lista de listas com campo_1 e campo_2 (se campo_1 >23)

# um update
with Session(engine) as session:
    result = session.execute(
        text("UPDATE tabela SET campo_1=:c1 WHERE campo_2=:c2"),
        [{"c1": 9, "c2": 11}, {"c1": 13, "c2": 15}],
    )
    session.commit()
[SQL]
UPDATE tabela SET campo_1=? WHERE campo_2=?
[...] [(9, 11), (13, 15)]

Vemos no exemplo que simplesmente substituimos as intruções:

with engine.connect() as conn  por  with Session(engine) as session
Connection.execute()           por  Session.execute()
Connection.commit()            por  Session.commit()

Obs.: Todas as consultas SQL são precedidas por BEGIN e terminadas por COMMIT (omitidas aqui).

Uma sessão com ORM


Para ilustrar o uso do ORM continuaremos, por enquanto, usando a construção de consultas com a função text("query"), que passa a string de consulta diretamente para o banco de dados. O artigo Sqlalchemy ORM Resumido contém uma amostra das funções básicas do ORM.

Começaremos com a tabela coordenadas do BD meu_banco.db do SQLite, construído na seção anterior. Na última operação ele foi gravado com o estado mostrado na figura. Faremos uma atualização de valores com UPDATE coordenadas SET y=:y WHERE x=:x. Os valores de :x, :y são lidos na lista de dicionários. Cada dicionário gera uma operação de UPDATE. Para conferir o resultado da atualização fazemos uma consulta somente dos valores com y > 100.

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

engine = create_engine("sqlite:///meu_banco.db")	

query = text("UPDATE coordenadas SET y=:y WHERE x=:x")
valores = [{"x": 11, "y": 110}, {"x": 15, "y": 150}]
with Session(engine) as session:
    result = session.execute(query, valores)
    session.commit()    

query = text("SELECT x, y FROM coordenadas WHERE y > :y ORDER BY x, y")
with Session(engine) as session:
    result = session.execute(query, {"y": 100})
    for row in result:
        print(f" x = {row.x}  y = {row.y}")
        
# a consulta com SELECT resulta em
↳  x = 11  y = 110
   x = 15  y = 150

As consultas geradas acima são, respectivamente:

[SQL]
UPDATE coordenadas SET y=110 WHERE x=11
UPDATE coordenadas SET y=150 WHERE x=15
SELECT x, y FROM coordenadas WHERE y > 100 ORDER BY x, y

A sessão não é tornada permamente (commited) automaticamente. Para isso é necessário emitir o comando session.commit(). Pelos exemplos mostrados vemos que simplesmente substituimos as intruções:

with engine.connect() as conn  por  with Session(engine) as session
Connection.execute()           por  Session.execute()
Connection.commit()            por  Session.commit()

Obs.: Todas as consultas SQL são precedidas por BEGIN e terminadas por COMMIT (omitidas aqui).

Definindo tabelas com ORM

Com o SQLAlchemy ORM temos uma sintaxe de criação de tabelas mais próxima do estilo do Python. Ele fornece uma interface chamada de Tabela Declarativa (Declarative Table) que usa tipos de variáveis do Python para representar e configurar as tabelas. Com esse procedimento temos classes mapeadas do Python que refletem as propriedades das tabelas do SQL. Em outras palavras, criamos classes do Python com atributos e propriedades que refletem tabelas, colunas, vínculos e relacionamentos que são mapeadas em tabelas do SQL. As operações CRUD usuais são feitas diretamente nos objetos que herdam dessas classes e que, depois, são transferidas para o BD.

A coleção MetaData é criada automaticamente (se uma não for explicitamente fornecida) e fica associada ao objeto chamado Base Declarativa (Declarative Base) que pode ser criado como instância da classe DeclarativeBase:

from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
    pass

# a coleção metadata é criada em Base
print(Base.metadata)
↳ MetaData()

Para definir nossas tabelas mapeadas herdamos de Base que, como vimos, herda de DeclarativeBase.

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

engine = create_engine("sqlite:///meu_banco.db")

class Base(DeclarativeBase):
    pass

class Aluno(Base):
    __tablename__ = "aluno"
    id: Mapped[int] = mapped_column(primary_key=True)
    matricula: Mapped[str] = mapped_column(String(50))
    nome: Mapped[str] = mapped_column(String(50))
    sobrenome: Mapped[Optional[str]]
    enderecos: Mapped[List["Endereco"]] = relationship(back_populates="aluno")
    
    def __repr__(self):
        return f"Aluno(id={self.id!r}, nome={self.nome!r}, sobrenome={self.sobrenome!r})"

class Endereco(Base):
    __tablename__ = "endereco"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str]
    aluno_id: Mapped[int] = mapped_column(ForeignKey("aluno.id"))    
    aluno: Mapped[Aluno] = relationship(back_populates="enderecos")

    def __repr__(self):
        return f"Aluno(id={self.id!r}, nome={self.nome!r}, sobrenome={self.sobrenome!r})"

Base.metadata.create_all(engine)

As duas classes, Alunos e Enderecos (e os objetos que herdam delas) ficam disponíveis para operações de persistência e consultas. Elas são denominadas classes mapeadas pelo ORM (ORM Mapped Classes). O nome de cada tabela fica atribuído em DeclarativeBase.__tablename__. Após a criação a tabela fica disponível por meio do atributo DeclarativeBase.__table__.

As colunas da tabela, por sua vez, são criadas por mapped_column() que usa anotações (a construção nome_campo: tipo_de_dado, que fica associado ao Mapped[tipo]. Se a coluna tem um tipo simples sem outras qualificações basta indicar apenas Mapped[tipo], onde os tipos do Python como int ou str significam as classes Integer ou String do SQLAlchemy, respectivamente. Essas definições podem ser bastante modificadas para representar objetos mais complexos.

Inserindo linhas

No ORM as instruções Insert são emitidas, e inseridas na transação✠, pelo objeto Session. Para isso inserimos novos objetos à Session e os tornamos persistentes (gravando a transação no BD) com com um processo chamado de flush. Esse processo é conhecido como padrão de unidade de trabalho (UoW)†.

Leia sobre Transações

Até agora inserimos dados usando INSERT com dicionários que contém as dados a serem incluídos. Com a abordagem ORM criamos novos objetos derivados das classes das classes customizadas que representam dados na tabela e os inserimos nos objetos table contidos em Session.

Vimos nos nossos exemplos como definir a estrutura de uma tabela criando classes de herdam de DeclarativeBase, criando uma classe para cada tabela SQL. Definimos as classes Aluno e Endereco e usamos Base.metadata.create_all(engine) para inserir no BD as tabelas representadas por elas. As mesmas classes são usadas na inserção de linhas.

Por exemplo, criamos abaixo dois objetos instâncias de Aluno, instanciando a classe e usando os nomes de colunas como keywords. Essa operação usa o construtor __init__() construído automaticamente pelo ORM.

jones = Aluno(matricula= '3456-1234', nome="Jones", sobrenome="Manoel", enderecos=[])
galileu = Aluno(matricula= '8888-9999',nome="Galileu", sobrenome="Galilei", enderecos=[])

# se exibirmos um objeto com print
print(jones)
↳ Aluno(id=None, matricula= '3456-1234', nome="Jones", sobrenome="Manoel", enderecos=[])

Observe que não incluimos um valor para id que é um campo de autoincremento inserido automaticamente. Se o objeto for exibido veremos que id=None, provisoriamente. Um valor é atribuído pelo mecanismo do banco de dados.

Objetos criados dessa forma são chamados transientes, pois não fazem parte ainda do BD, nem mesmo da representação no ORM. Essa inserção deve ser feita na Session, com o método .add(). Feito isso as linhas são pendentes, ainda não inseridas no BD. Esse estado pode ser verificado por meio do objeto Session.new.

# criamos uma sessão
session = Session(engine)

# inserimos os objetos (que representam linhas) na sessão
session.add(jones)
session.add(galileu)

# para verificar objetos pendentes
print(session.new)
↳ IdentitySet([Aluno(id=None, matricula= '3456-1234', nome="Jones", sobrenome="Manoel"),
               Aluno(id=None, matricula= '8888-9999',nome="Galileu", sobrenome="Galilei")]

# para inserir esses valores no modelo do BD
session.flush()
[SQL]
INSERT INTO aluno (matricula, nome, sobrenome) VALUES (?, ?), (?, ?) RETURNING id
('3456-1234', 'Jones', 'Manoel', '8888-9999', 'Galileu', 'Galilei')

A consulta realizada insere os dois objetos criados na Session e retorna os ids das linhas inseridas. Para isso é usado o padrão de unidade de trabalho (UoW), o que significa que as alterações não são comunicadas ao BD até que o método Session.flush() seja usado. A transação aberta no início com Session(engine) permanece aberta até que sejam emitidos um dos comandos, chamando métodos de Session:

Session.commit()
Session.rollback() # ou
Session.close()

A execução de .commit() também emite um .flush(). É possível configurar uma Session para que o comportamento autoflush (flush automático).

Recuperando pks: Quando um objeto é inserido o ORM gera automaticamente os atributos das chaves primárias (pk). Os objetos criados acima, jones e galileu passam a ter um id que pode ser lidos.

print(jones.id)
↳ 4
print(galileu.id)
↳ 5

Essa propriedade é lida internamente com CursorResult.inserted_primary_key e esse procedimento exige que operações de INSERT sejam feitas uma de cada vez. Por isso não foram feitas operações tipo executemany. Alguns gerenciadores, como o psycopg2 do PostgreSQL, são capazes de inserir várias linhas de uma vez e recuperar suas chaves primárias.

‡ Um mapa de identidade (identity map) é um mapeamento entre objetos do Python os objetos (tabelas, linhas, colunas, etc) representados no banco de dados. Ele é uma coleção mantida na memória no objeto Session do ORM que contém objetos relacionados por meio de suas chaves primárias. Esse padrão permite que todas as operações sobre o BD sejam coordenadas por em uma única instância de objeto. Veja: Martin Fowler, Identity Map.

Podemos recuperar um dos objetos armazenados no mapa de identidade‡ usando o método Session.get()

um_aluno = session.get(Aluno, 4)
print(um_aluno)
↳ Aluno(id=None, matricula= '3456-1234', nome="Jones", sobrenome="Manoel", enderecos=[])

# o objeto é o mesmo que o definido anteriormente
um_aluno is jones
↳ True

O objeto é retornado se existir no mapa ou, caso contrário, um SELECT é produzido. Observamos que get() retorna uma referência para o mesmo objeto já existente (desde que não tenha sido removido).

Committing: Comitar o estado do BD significa gravar as alterações feitas no BD. Após o commit, todos os objetos continuam ligados (attached) à seção até que ela seja encerrada. Se estamos usando um gerenciador de contexto, abrindo e usando a sessão dentro de um bloco with, a sessão é fechada ao abandonarmos o bloco. Caso contrário temos que fechar manualmente a sessão com session.close().

# para "comitar" uma sessão
session.commit()
[SQL]
COMMIT

# para fechar a sessão
session.close()

Update com UoW: Suponha que desejamos alterar a linha da tabela Aluno referente ao aluno de nome Galileu (que assumiremos tem id=4). Primeiro carregamos essa linha em um objeto (caso já não esteja carregada).

galileu = session.execute(select(Aluno).filter_by(name="Galileu")).scalar_one()
[SQL]
SELECT aluno.id, aluno.nome, aluno.sobrenome, aluno.enderecos
FROM aluno WHERE alunou.nome = ? ('Galileu',)

# o objeto é criado com os dados da linha
print(galileu)
galileu = Aluno(id=4, matricula= '8888-9999',nome="Galileu", sobrenome="Galilei", enderecos='')

Podemos alterar propriedades desse objeto da forma usual. A alteração fica armazenada em session, inicialmente na coleção chamada session.dirty, que contém objetos alterados antes de flush. Após session.flush() um UPDATE é executado no BD e o objeto alterado sai da coleção session.dirty (lembrando que podemos ajustar um autoflush).

galileu.sobrenome = "Osbourne"
galileu in session.dirty
↳ True

session.flush()
galileu in session.dirty
↳ False

# após a operação o novo dado pode ser verificado

galileu_sobrenome = session.execute(select(Aluno.sobrenome)
                           .where(Aluno.id == 4)).scalar_one()
print(galileu_sobrenome)
↳ Osbourne

# as linhas correspondem a
[SQL]
UPDATE aluno SET sobrenome=? WHERE aluno.id = ? ('Osbourne', 4)
SELECT aluno.sobrenome FROM aluno WHERE aluno.id = ? (4,)

Um flush é executado quando SELECT é executado.

Deleting com UoW: Uma linha pode ser removida do BD com Session.delete(obj), onde obj é um objeto carregado representando essa linha. O objeto permanece na sessão até a emissão de um flush e, depois, é removido dela. Da mesma forma que ocorre com a consulta de UPDATE, as alterações do estado do BD só é permanente quando se realiza um commit.

jones = session.get(Aluno, 1)
session.delete(jones)

# após um flush
session.flush()   # ou uma consulta de SELECT emitida
jones in session
↳ False

# para a permanência no BD
session.commit()

Operações de INSERT, UPDATE e DELETE em várias linhas: Vimos que objetos são inseridos em uma sessão com Session.add() e o mecanismo interno da ORM cuida da emissão de consultas SQL relacionadas.

Além dessa funcionalidade, sessões ORM também podem processar instruções INSERT, UPDATE e DELETE diretamente sem passar pela criação de outros objetos da ORM, recebendo listas de valores a serem inseridas, atualizados ou apagados, incluindo critérios WHERE que aplicam a transformação em muitas linhas de uma vez. Isso é útil quando se quer aplicar a alteração a muitas linhas, evitando a construção de objetos mapeados.

As sessões ORM podem se utilizar de recursos com insert(), update() e delete() de forma similar à usada no CORE. Para isso essas funções recebem coleções em seus argumentos, em geral uma lista de dicionários. Por exemplo:

from sqlalchemy import insert
session.execute(
    insert(Aluno),
    [
        {"matricula":"9487634", "nome": "Rodrigo", "sobrenome": "Santos"},
        {"matricula":"0698734", "nome": "Paula", "sobrenome": "Silva"},
        {"matricula":"9998765", "nome": "Humberto", "sobrenome": "Loyola"},
        {"matricula":"1230984", "nome": "Mariane", "sobrenome": "Louise"},
        {"matricula":"2345670", "nome": "Afonso", "sobrenome": "Pena"},
    ],
)
[SQL]
INSERT INTO aluno (matricula, nome, sobrenome) VALUES (?, ?, ?)
[
  ('9487634', 'Rodrigo' 'Santos'),
  ('0698734', 'Paula',  'Silva'),
  ('9998765', 'Humberto', 'Loyola'),
  ('1230984', 'Mariane',  'Louise'),
  ('2345670', 'Afonso', 'Pena')
]

Outras informações no Guia do SQLAlchemy: ORM-Enabled INSERT, UPDATE, and DELETE statements.

Desfazendo transações com Roll Back: Uma sessão possui o método Session.rollback() que se destina a emitir um ROLLBACK na conexão SQL ativa. Esse método também afeta os objetos associados à Session, como é o caso do objeto galileu armazenada em nossos exemplos. Fizemos no exemplo a alteração da propriedade galileu.sobrenome de "Galilei" para "Osbourne". Se aplicarmos Session.rollback() toda a transação atual será cancelada e todos os objetos em associação com a sessão ficarão expirados.

Fechando uma sessão: Em vários dos exemplos usados abrimos e manipulamos as sessões fora de um gerenciador de contexto e, portanto, elas devem ser fechadas manualmente. Claro que a alternativa é usar o gerenciador, como mostrado abaixo.

from sqlalchemy.orm import Session
engine = create_engine("url/do/banco_de_dados")	

session = Session(engine)
#  conjunto de operações sobre o BD
session.commit()
session.close()


# usando o gerenciador de contexto
with Session(engine) as session:
    # conjunto de operações sobre o BD
    session.commit()

Quando fechamos uma sessão, manualmente ou por meio de um gerenciador de contexto, liberamos os recursos de máquina usados para a conexão. Se existirem transações não comitadas elas serão perdidas (emitindo um ROLLBACK). Portanto, se usarmos a sessão apenas para operações de leitura, como em SELECTs, basta fechá-la, sem preocupação com a emissão de Session.rollback(). Além disso todos os objetos ligados à sessão são desconectados, ficando sujeitos à execução de limpeza pelo gc, coletor de lixo do Python. Veja Python Manual: Garbage Collector interface.

Transações

Retornar para o artigo principal

Uma transação é a menor unidade de operações realizadas sobre um banco de dados. Ela é composta de um conjunto ordenado de instruções e pode ser executada manualmente ou automatizada no código. Os sistemas gerenciadores devem garantir que a transação seja executada por completo ou abandonada, sem nenhuma alteração ao banco.

†: Padrão de Unidade de Trabalho (Unit Of Work, UoW) Unidade de Trabalho é um padrão de projeto onde se mantém uma lista de objetos afetados por uma transação e coordena como essas alterações são efetivadas, cuidando de possíveis problemas de concorrência. O padrão Unit of Work pode ser visto como um contexto, sessão ou objeto que acompanha as alterações das entidades de negócio durante uma transação e está presente em muitas das ferramentas ORM modernas. O objetivo das UoW é o agrupamento de funções e alterações aplicados sobre um banco de dados que possa ser executado de uma vez, ou abandonado por completo. Veja artigo de Martin Fowler.

COMMIT e ROLLBACK: Um COMMIT é a instrução para efetivar, tornando permanentes, as operações sobre o BD desde que o último COMMIT ou ROLLBACK foi feito. Um ROLLBACK é a instrução SQL usado para reverter o estado do BD para o estado tornado efetivo pela última operação COMMIT ou ROLLBACK.

Se uma transação for concluída com sucesso o banco de dados será alterado permanentemente, com gravação em disco dos dados alterados, na operação de COMMIT. Porém, se houver falha em qualquer uma das operações da transação, o banco deve ser deixado em seu estado inicial, coom um ROLLBACK.

Transações devem possuir início e fim e podem ser salvas (permanência no banco de dados) ou desfeitas. Se houver falha nenhuma operação deve ser tornada permanente.

No SQL transações são iniciadas com BEGIN TRANSACTION, e finalizada com COMMIT ou ROLLBACK. Essas operações estão ilustradas abaixo.

-- criamos uma tabela provisória de testes
SELECT matricula, nome INTO temp_aluno FROM aluno;

-- transação com rollback
BEGIN TRANSACTION
  DELETE FROM temp_aluno        -- apaga todos registros da tabela
  SELECT * FROM temp_aluno      -- a tabela está vazia
ROLLBACK TRANSACTION;           -- desfaz a transação
SELECT * FROM temp_aluno;       -- a tabela está como no início
                               
-- transação com commit        
BEGIN TRANSACTION              
  DELETE FROM temp_aluno        -- apaga todos registros da tabela
  SELECT * FROM temp_aluno      -- a tabela está vazia
COMMIT TRANSACTION;             -- confirma a transação
SELECT * FROM temp_aluno;       -- a tabela está vazia (permanente)

Fonte: Boson Treinamentos: Transacões, commit e rollback.

Bibliografia

Esse texto é baseado primariamente na documentação do SQLAlchemy, disponível em SQLAlchemy 2, Documentation. Outras referências no artigo Python e SQL: SQLAlchemy.

SQLAlchemy – ORM (Exemplo de Uso)

Posted on by Guilherme Silva


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.