Entendo e resolvendo o problema de N+1

N+1

É uma situação ao procurar dados relacionados (ex: os livros do autor que já foi encontrado). A primeira busca traz N itens (autores), e o código faz mais N buscas adicionais para os dados relacionados (uma para os livros de cada autor). Ou seja N buscas (livros de cada autor) + 1 busca (autores)

Exemplo: Criei um banco de dados simples chamado livraria, com as tabelas autor (nome, id, qtd_livros) e livros (id, titulo, id_autor e data_lancamento).

Com o banco feito, crie um ORM em Python usando o SQLAlchemy:

Python

from datetime import date
from sqlalchemy import create_engine, ForeignKey, String, Integer, Date
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker

# 1. Configuração da Conexão
USER = "root"
PASSWORD = ""
HOST = "localhost"
DATABASE = "livraria"
DATABASE_URL = f"mysql+pymysql://{USER}:{PASSWORD}@{HOST}/{DATABASE}"

# O 'engine' é o motor que gerencia a conexão física com o MySQL
# 'echo=True' faz o SQLAlchemy mostrar no terminal o SQL real que ele está gerando (ótimo para aprender)
engine = create_engine(DATABASE_URL, echo=True)

# 2. Configuração da Sessão (responsável por abrir e fechar conversas com o banco)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 3. Classe Base que todos os nossos modelos vão herdar
class Base(DeclarativeBase):
  pass

# 4. Mapeamento da Tabela 'autor'
class Autor(Base):
  __tablename__ = "autor"
  
  # Mapped[...] define o tipo no Python e mapped_column define as regras do Banco de Dados
  id: Mapped[int] = mapped_column(Integer, primary_key=True)
  nome: Mapped[str] = mapped_column(String(100), nullable=False)
  qtd_livros: Mapped[int] = mapped_column(Integer, default=0)

  # Relacionamento virtual: Permite acessar os livros do autor direto pelo Python (ex: autor.livros)
  livros: Mapped[list["Livro"]] = relationship(back_populates="autor")

# 5. Mapeamento da Tabela 'livros'
class Livro(Base):
  __tablename__ = "livros"

  id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
  titulo: Mapped[str] = mapped_column(String(150), nullable=False)
  autor_id: Mapped[int] = mapped_column(Integer, ForeignKey("autor.id"), nullable=False)
  data_lancamento: Mapped[date] = mapped_column(Date, nullable=False)

  # Relacionamento virtual inverso: Permite saber quem é o autor do livro (ex: livro.autor.nome)
  autor: Mapped["Autor"] = relationship(back_populates="livros")

E um script para fazer a consulta com um contador de quantas consultas foram feitas:

Python

from sqlalchemy import event
from sqlalchemy.orm import joinedload
from banco import engine, SessionLocal, Autor

# 1. Um pequeno objeto para armazenar um contador de quantas consultas foram feitas
class ContadorDeQueries:
  quantidade = 0

@event.listens_for(engine, "before_cursor_execute")
def contar_queries(conn, cursor, statement, parameters, context, executemany):
  ContadorDeQueries.quantidade += 1

session = SessionLocal()

print("Iniciando as consultas...\n")

# Zerar o contador
ContadorDeQueries.quantidade = 0

# essa consulta causa o N+1 > autores = session.query(Autor).all() mas por que?
# a consulta session.query(Autor).all(), carrega só as informações dos autores na memória, se as informações dos livros precisarem ser carregadas, o programa terá que fazer uma consulta nova para cada conjunto de livros
autores = session.query(Autor).options(joinedload(Autor.livros)).all() # já essa consulta carrega os autores e os livros de uma vez só

# Iteramos sobre a lista de autores
for autor in autores:
  print(f"Autor: {autor.nome}")
  # Acessamos os livros (isso vai disparar novas consultas e acionar nosso evento)
  for livro in autor.livros:
      print(f"  - Livro: {livro.titulo}")

# Exibir o resultado do contador no final
print("\n" + "="*40)
print(f"TOTAL DE CONSULTAS FEITAS: {ContadorDeQueries.quantidade}")
print("="*40 + "\n")

session.close()

Mas ainda tem um problema crítico com essa consulta, o autores = session.query(Autor).options(joinedload(Autor.livros)).all(), carrega todos os registros na RAM de uma vez, se o banco retornar 1 milhão de registros, isso vai travar uma máquina mais fraca. Para resolver isso existem algumas formas:

Paginação:

Python

pagina_atual = 1 
itens_por_pagina = 50 
deslocamento = (pagina_atual - 1) * itens_por_pagina 

autores = session.query(Autor)\
  .options(joinedload(Autor.livros))\
  .limit(itens_por_pagina)\
  .offset(deslocamento)\
  .all()
# Puxa os registros em pequenas partes de 50

Processamento em lote:

Python

query = session.query(Autor) 
for autor in query.yield_per(1000):  
  faz_algum_calculo(autor)
# Puxa 1000 registros, e quando puxar os próximos 1000, os antigos são descartados da memória

Consulta com filtro (WHERE):

Python

autores_recentes = session.query(Autor).filter(Autor.ano_publicacao == 2023).all()

Fontes e créditos: https://planetscale.com/blog/what-is-n-1-query-problem-and-how-to-solve-it