PyMongo Essencial: Simplificando a Conexão entre MongoDB e Python

Simplifique as Operações com MongoDB usando Python: Um Guia para Iniciantes sobre Inserção, Consulta e Agregação de Dados de Forma Eficiente para Seus Projetos Orientados a Dados

🕒 Tempo de leitura estimada: 16 minutos

Na era do big data, ter uma maneira poderosa e flexível de gerenciar suas operações de banco de dados é indispensável. MongoDB é um banco de dados NoSQL altamente escalável, conhecido por sua velocidade e facilidade de uso. Juntamente com a linguagem de programação Python, você tem uma combinação poderosa para projetos dinâmicos e orientados a dados. PyMongo é a biblioteca preferida para conectar MongoDB com Python. Aqui, iremos guiá-lo pelos recursos básicos, intermediários e avançados do PyMongo, enriquecidos com exemplos práticos.

Agora, vamos mergulhar no mundo do MongoDB em Python com alguns exemplos práticos que também estão disponíveis no Google Colab aqui 👨‍🔬.

Introdução ao PyMongo

PyMongo é o driver oficial de Python para MongoDB, permitindo que você interaja perfeitamente com bancos de dados MongoDB. Quer você esteja trabalhando em um script simples ou em uma aplicação complexa orientada a dados, o PyMongo fornece um conjunto de ferramentas abrangente para todas as suas operações de banco de dados.

Antes de mergulhar nas diversas operações, você precisa instalar o PyMongo:

pip install pymongo

Operações Básicas

Conectando ao MongoDB

Primeiro, vamos conectar a um servidor MongoDB local. Neste exemplo, estamos usando uma instância gratuita do MongoDB Atlas fornecida pela InfinitePy. É importante notar que o username e o password estão hard-coded neste exemplo. Por motivos de segurança, em aplicativos reais, essas credenciais devem ser gerenciadas usando variáveis de ambiente ou gerenciadores de segredos.

# Importa a classe MongoClient da biblioteca pymongo.
# Isso nos permitirá conectar a um banco de dados MongoDB.
from pymongo import MongoClient

# Define o nome de usuário para o banco de dados MongoDB. Isso deve ser um usuário válido com permissões apropriadas.
username = 'infinitepy'

# Define a senha para o banco de dados MongoDB. Assegure que esta senha esteja correta.
# Nunca exponha ou hard-code senhas no seu código por motivos de segurança. Use variáveis de ambiente ou gerenciadores de segredos em aplicativos reais.
password = 'ilovepyth0n'

# Cria um objeto MongoClient. Este objeto representa uma conexão de cliente com o banco de dados MongoDB.
# Usamos uma f-string para formatar a string de conexão com o nome de usuário e a senha.
# O URI de conexão inclui o nome de usuário, senha e informações do cluster.
client = MongoClient(f'mongodb+srv://{username}:{password}@infinitepy-cluster.s05fljv.mongodb.net/?retryWrites=true&w=majority&appName=infinitepy-cluster')

Notas sobre os parâmetros do URI de conexão:

  • mongodb+srv: Especifica o protocolo a ser usado para conectar ao cluster MongoDB por meio de uma string de conexão 'SRV'.

  • {username}:{password}: Substitui os placeholders pelo nome de usuário e senha reais. ⚠️ Garanta que suas credenciais estejam seguras e protegidas.

  • ?retryWrites=true: Garante que operações de escrita sejam repetidas uma vez se falharem devido a um erro de rede.

  • &w=majority: Garante que operações de escrita só retornem com sucesso uma vez que a maioria dos nós do cluster reconheçam.

  • &appName=infinitepy-cluster: Especifica um nome personalizado para esta conexão. Útil para fins de monitoramento.

Listar todos os nomes de bancos de dados existentes de um cliente MongoDB

# Usando a instância do cliente, obtenha a lista de todos os nomes de bancos de dados
# list_database_names() retorna uma lista de strings, onde cada string é o nome de um banco de dados.
for db_name in client.list_database_names():
    # Imprime cada nome de banco de dados no console
    print(db_name)

Executar o código acima produzirá a seguinte saída:

library
sample_db
sample_mflix
admin
local

Inserindo um Documento

Uma das operações mais básicas é inserir documentos em uma coleção. Aqui está um exemplo simples:

# Acesse o banco de dados sample_db; se não existir, o MongoDB o cria quando você armazena algum dado nele pela primeira vez.
# A sintaxe 'client.sample_db' é usada para acessar o banco de dados.
db = client.sample_db

# Acesse a coleção chamada 'minha_colecao' no banco de dados sample_db
collection = db.minha_colecao

# Crie um dicionário representando o documento a ser inserido na coleção
# O documento contém informações sobre uma pessoa: nome, idade e cidade
documento = {"nome": "João Silva", "idade": 29, "cidade": "São Paulo"}

# Insira o documento na coleção 'minha_colecao'
# O método insert_one retorna um objeto result que contém informações sobre a inserção
result = collection.insert_one(documento)

# Imprime o ID único do documento inserido
# O atributo inserted_id do objeto result contém esse ID
print("ID do documento inserido: {document_id}".format(document_id=result.inserted_id))

Consultando Documentos

Para recuperar documentos, você pode usar o método find:

# Acesse o banco de dados sample_db; se não existir, o MongoDB o cria quando você armazena algum dado nele pela primeira vez.
# A sintaxe 'client.sample_db' é usada para acessar o banco de dados.
db = client.sample_db

# Acesse a coleção chamada 'minha_colecao' no banco de dados sample_db.
# Uma coleção no MongoDB é semelhante a uma tabela em um banco de dados relacional.
collection = db.minha_colecao

# Encontre um único documento na coleção 'minha_colecao' onde o campo 'nome' seja igual a "João Silva".
# O método 'find_one' retorna a primeira correspondência encontrada.
result = collection.find_one({"nome": "João Silva"})

# Imprime o resultado no console.
# Se um documento com o 'nome' "João Silva" for encontrado, imprimirá esse documento.
# Se nenhum documento assim existir, imprimirá 'None'.
print(result)

Executar o código acima produzirá a seguinte saída.

{'_id': ObjectId('666a1d2d23a1c5436afa5f1a'), 'nome': 'João Silva', 'idade': 29, 'cidade': 'São Paulo'}

⚠️ O ObjectId será diferente para cada novo documento criado.

Inserindo múltiplos Documentos

Você pode automatizar a inserção de múltiplos documentos a partir de uma lista:

# Acessar o banco de dados sample_db; o MongoDB o cria se não existir.
db = client.sample_db

# Acessar a coleção chamada 'minha_colecao' dentro do banco de dados sample_db.
colecao = db.minha_colecao

# Define uma lista de documentos para serem inseridos em 'minha_colecao'.
# Cada documento é representado como um dicionário Python com pares de chave-valor.
documentos = [
    {"nome": "Alice", "idade": 25, "cidade": "Londres"},
    {"nome": "Bob", "idade": 30, "cidade": "São Francisco"},
]

# Insere a lista de documentos na coleção.
# O método insert_many insere múltiplos documentos de uma vez e retorna um objeto contendo os inserted_ids.
resultado = colecao.insert_many(documentos)

# Imprimir os IDs únicos dos documentos inseridos.
print("Documentos inseridos com IDs:", resultado.inserted_ids)

Atualizando Documentos

Em aplicações do mundo real, você frequentemente precisará atualizar registros existentes. Veja como isso pode ser feito:

# Acessar o banco de dados 'sample_db'; se ele não existir, o MongoDB o cria quando você armazena algum dado nele.
db = client.sample_db

# Acessar a coleção chamada 'minha_colecao' no banco de dados 'sample_db'.
colecao = db.minha_colecao

# Executar a operação de atualização. O método 'update_one' encontra o primeiro documento que corresponde ao filtro e
# atualiza esse documento.
resultado = colecao.update_one(
    {"nome": "João da Silva"},  # Filtro: encontra o documento onde o nome é 'João da Silva'
    {"$set": {"idade": 30}}     # Atualização: define a idade como 30
)

# Verifica se algum documento foi modificado.
if resultado.modified_count > 0:
    print("Documento atualizado com sucesso.")  
else:
    print("Nenhum documento foi atualizado.")   

Deletando Documentos

Remover documentos também é algo simples:

# Acessar o banco de dados sample_db; o MongoDB o cria se não existir.
# A sintaxe 'client.sample_db' é usada para acessar o banco de dados.
db = client.sample_db

# Acessar a coleção chamada 'minha_colecao' dentro do banco de dados sample_db.
# Uma coleção no MongoDB é similar a uma tabela em um banco de dados relacional.
colecao = db.minha_colecao

# Tentar deletar um único documento da coleção 'minha_colecao'
# onde o campo 'nome' é "João da Silva".
# O método `delete_one` deleta o primeiro documento que corresponder aos critérios do filtro.
resultado = colecao.delete_one({"nome": "João da Silva"})

# Verifica se um documento foi realmente deletado examinando o atributo 'deleted_count'.
# Esse atributo retorna o número de documentos que foram deletados.
if resultado.deleted_count > 0:
    # Se deleted_count for maior que 0, significa que pelo menos um documento foi deletado.
    print("Documento deletado com sucesso.")
else:
    # Se deleted_count for 0, significa que nenhum documento correspondeu aos critérios do filtro.
    print("Nenhum documento foi deletado.")

Operações Avançadas

Framework de Agregação

O Framework de Agregação do MongoDB é uma poderosa ferramenta para processar dados e transformar documentos dentro de uma coleção. Em vez de usar código imperativo para processar dados, você pode definir um pipeline de operações que o MongoDB executará. Cada estágio no pipeline transforma os documentos à medida que eles passam. O framework de agregação é particularmente útil para tarefas como filtragem, agrupamento, cálculo de médias e outros tipos de processamento de dados.

Conceitos e Estágios Principais

  • Pipeline: Uma sequência de estágios. Os documentos entram no pipeline e são processados sequencialmente por cada estágio.

  • Estágio: Cada estágio opera nos documentos e pode realizar operações como filtragem, agrupamento, projeção, ordenação e mais.

  • Estágios Comuns:

    • $match: Filtra documentos por uma condição especificada. Similar à cláusula WHERE no SQL.

    • $group: Agrupa documentos por uma chave especificada e pode calcular valores agregados como somas, médias e contagens.

    • $project: Remodela documentos, incluindo ou excluindo campos ou calculando novos campos.

    • $sort: Ordena documentos com base em um ou mais campos.

    • $limit e $skip: Controlam o número de documentos a retornar, permitindo paginação.

No exemplo de código fornecido:

  1. Etapa 1 ($match): O pipeline começa com a filtragem de documentos. Nesse caso, apenas documentos em que o campo idade é maior ou igual a 25 são passados ​​para o próximo estágio.

  2. Etapa 2 ($group): Os documentos filtrados são agrupados pelo campo cidade (_id é definido como $cidade). Para cada cidade, a idade média é calculada usando o operador acumulador $avg, e o resultado é armazenado no campo media_idade.

Este pipeline de duas etapas calcula efetivamente a idade média de todas as pessoas com 25 anos ou mais para cada cidade na coleção.

# Acessar o banco de dados sample_db; o MongoDB o cria se não existir.
# A sintaxe 'client.sample_db' é usada para acessar o banco de dados.
db = client.sample_db

# Acessar a coleção chamada 'minha_colecao' dentro do banco de dados sample_db.
# Uma coleção no MongoDB é similar a uma tabela em um banco de dados relacional.
colecao = db.minha_colecao

# Define um pipeline de agregação para processar documentos na coleção.
# O pipeline é uma lista de estágios, onde cada estágio transforma os documentos à medida que passam.
# Os estágios são processados em ordem, do primeiro ao último.
pipeline = [
    # O estágio $match filtra os documentos pela condição especificada.
    # Aqui, estamos interessados apenas em documentos onde o campo idade seja maior ou igual a 25.
    {"$match": {"idade": {"$gte": 25}}},

    # O estágio $group agrupa os documentos por um identificador especificado.
    # Aqui, estamos agrupando documentos pelo campo 'cidade'.
    # O campo '_id' é usado como o identificador do grupo.
    # O campo 'media_idade' é calculado como o valor médio do campo 'idade' para cada grupo.
    {"$group": {"_id": "$cidade", "media_idade": {"$avg": "$idade"}}},
]

# Executa o pipeline de agregação na coleção.
# O método aggregate() retorna um iterável com os resultados do pipeline.
resultado = colecao.aggregate(pipeline)

# Itera sobre o resultado e imprime cada documento.
# Cada documento no resultado representa uma cidade e sua respectiva idade média.
for doc in resultado:
    print(doc)

Executar o código acima produzirá a seguinte saída:

{'_id': 'São Paulo', 'media_idade': 29.0}
{'_id': 'Londres', 'media_idade': 25.0}
{'_id': 'São Francisco', 'media_idade': 30.0}

Exemplo Prático

Criando um pipeline de agregação complexo para encontrar a idade máxima por cidade:

# Acessar o banco de dados sample_db; o MongoDB o cria se não existir.
# A sintaxe 'client.sample_db' é usada para acessar o banco de dados.
db = client.sample_db

# Acessar a coleção chamada 'minha_colecao' dentro do banco de dados sample_db.
# Uma coleção no MongoDB é similar a uma tabela em um banco de dados relacional.
colecao = db.minha_colecao

# Define um pipeline de agregação para processar os dados na coleção.
# O pipeline contém uma sequência de estágios, cada um representado por um documento.
pipeline = [
    # O estágio $group agrupa documentos pelo campo 'cidade'.
    # Ele cria um novo documento para cada grupo onde '_id' é o nome da cidade.
    # Ele também calcula a idade máxima para cada grupo e armazena em 'idade_max'.
    {"$group": {"_id": "$cidade", "idade_max": {"$max": "$idade"}}},

    # O estágio $sort ordena os documentos agrupados pelo campo 'idade_max' em ordem decrescente (-1).
    {"$sort": {"idade_max": -1}}
]

# Execute o pipeline de agregação na coleção.
# 'colecao.aggregate(pipeline)' retorna um cursor para o resultado da agregação.
resultado = colecao.aggregate(pipeline)

# Itera sobre os documentos no cursor de resultado.
# Cada documento em 'resultado' é um dicionário contendo '_id' (o nome da cidade) e 'idade_max'.
for doc in resultado:
    # Imprimir o nome da cidade e a idade máxima para cada grupo.
    print(f"Cidade: {doc['_id']}, Idade Máxima: {doc['idade_max']}")

Executar o código acima produzirá a seguinte saída:

Cidade: São Francisco, Idade Máxima: 30
Cidade: São Paulo, Idade Máxima: 29
Cidade: Londres, Idade Máxima: 25

Projeto em Ação: Da Teoria à Prática

Agora que cobrimos o básico, vamos pôr nosso conhecimento em prática criando uma aplicação simples para gerenciar uma biblioteca de livros. Este projeto demonstrará operações CRUD (Criar, Ler, Atualizar, Deletar) fundamentais usando o MongoDB com a ajuda do pymongo, mostrando esses conceitos em um cenário do mundo real. Além disso, implementaremos as técnicas de logging em Python discutidas em nosso último texto aqui.

import logging

# Criar um objeto logger com o nome 'mongo_logger'
mongo_logger = logging.getLogger('mongo_logger')

# Definir o nível de registro como DEBUG. Isso significa que todas as mensagens com nível DEBUG e acima serão registradas
mongo_logger.setLevel(logging.DEBUG)

# Criar manipuladores que imprimem mensagens de log no console (saída padrão) e escrevem mensagens de log em um arquivo chamado 'mongo_logger.log'
console_handler = logging.StreamHandler()
file_handler = logging.FileHandler('mongo_logger.log')

# Definir um formatador que especifica o formato das mensagens de log
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# Definir o formatador para o manipulador do console e para o manipulador de arquivos
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)

# Adicionar o manipulador do console ao logger. Isso significa que mensagens de log serão exibidas no console
mongo_logger.addHandler(console_handler)
# Adicionar o manipulador de arquivos ao logger. Isso significa que mensagens de log serão escritas no arquivo 'mongo_logger.log'
mongo_logger.addHandler(file_handler)
# Função para obter um banco de dados específico
def get_database(db_name):
    # Registrar que estamos acessando o banco de dados especificado
    mongo_logger.info(f"Acessando o banco de dados '{db_name}'.")
    # Retornar o objeto do banco de dados do cliente
    return client[db_name]

# Função para obter uma coleção específica de um banco de dados
def get_collection(db, collection_name):
    # Registrar que estamos acessando a coleção especificada dentro do banco de dados fornecido
    mongo_logger.info(f"Acessando a coleção '{collection_name}' dentro do banco de dados '{db.name}'.")
    # Retornar o objeto da coleção do banco de dados
    return db[collection_name]

# Função para inserir múltiplos documentos de livros em uma coleção
def insert_books(collection, books):
    # Registrar que estamos inserindo novos documentos de livros
    mongo_logger.info("Inserindo novos documentos de livros na coleção.")
    # Inserir a lista de livros na coleção usando insert_many
    result = collection.insert_many(books)
    # Registrar os IDs dos documentos inseridos
    mongo_logger.info(f'IDs dos livros inseridos: {result.inserted_ids}')
    # Retornar o resultado da operação de inserção
    return result

# Função para encontrar livros por um autor específico em uma coleção
def find_books_by_author(collection, author_name):
    # Registrar que estamos consultando por livros do autor especificado
    mongo_logger.info(f"Consultando a coleção por todos os livros do autor {author_name}.")
    # Consultar a coleção por livros com o nome do autor fornecido
    books = collection.find({"author": author_name})
    # Registrar o número de livros encontrados (converter o cursor para lista para contar)
    mongo_logger.info(f'Encontrados {len(list(books))} livros do autor {author_name}.')
    # Retornar o cursor para a lista de livros encontrados
    return books

# Função para atualizar o gênero de um livro pelo seu título
def update_book_genre(collection, title, new_genre):
    # Registrar que estamos atualizando o gênero do livro especificado
    mongo_logger.info(f"Atualizando o gênero de '{title}' para '{new_genre}'.")
    # Atualizar o gênero do livro com o título fornecido
    update_result = collection.update_one(
        {"title": title},
        {"$set": {"genre": new_genre}}
    )
    # Registrar o número de documentos correspondentes e modificados
    mongo_logger.info(f'Modificados {update_result.matched_count} documento(s) e {update_result.modified_count} documento(s).')
    # Retornar o resultado da operação de atualização
    return update_result

# Função para deletar um livro pelo seu título
def delete_book(collection, title):
    # Registrar que estamos deletando o livro com o título especificado
    mongo_logger.info(f"Deletando o livro com o título '{title}'.")
    # Deletar o documento do livro com o título fornecido
    delete_result = collection.delete_one({"title": title})
    # Registrar o número de documentos deletados
    mongo_logger.info(f'Deletados {delete_result.deleted_count} documento(s).')
    # Retornar o resultado da operação de deleção
    return delete_result
# O script a seguir executa operações básicas em um banco de dados da biblioteca.
# Ele assume a existência de certas funções auxiliares: get_database, get_collection,
# insert_books, find_books_by_author, update_book_genre e delete_book.

if __name__ == "__main__":

    # Inicializar a conexão com o banco de dados com o nome 'biblioteca'
    db = get_database('biblioteca')

    # Acessar a coleção 'livros' do banco de dados 'biblioteca'
    livros_collection = get_collection(db, 'livros')

    # Definir uma lista de novos livros a serem adicionados à coleção
    novos_livros = [
        {
            "titulo": "O Apanhador no Campo de Centeio",
            "autor": "J.D. Salinger",
            "ano": 1951,
            "genero": "Ficção"
        },
        {
            "titulo": "O Sol é para Todos",
            "autor": "Harper Lee",
            "ano": 1960,
            "genero": "Ficção"
        },
        {
            "titulo": "Python Crash Course",
            "autor": "Eric Matthes",
            "ano": 2015,
            "genero": "Programação"
        }
    ]

    # Inserir a lista de novos livros na coleção 'livros'
    insert_books(livros_collection, novos_livros)

    # Encontrar todos os livros do autor J.D. Salinger na coleção 'livros'
    salinger_livros = find_books_by_author(livros_collection, "J.D. Salinger")

    # Registrar os livros encontrados pelo autor J.D. Salinger
    mongo_logger.info(f'Livros encontrados: {list(salinger_livros)}')

Conclusão

Integrando PyMongo em seus projetos Python, você pode aproveitar o poder do MongoDB de forma fácil. Desde inserções básicas e consultas até agregações avançadas e indexação, o PyMongo oferece um conjunto robusto de ferramentas para simplificar suas operações de banco de dados. Seja você um desenvolvedor Python iniciante, intermediário ou avançado, este guia fornece uma base abrangente para aproveitar todo o potencial do PyMongo.

Fique à vontade para responder a esta newsletter com quaisquer perguntas ou tópicos que gostaria que abordássemos no futuro.

Se gostou desta newsletter, não se esqueça de se inscrever para receber atualizações regulares. Compartilhe com seus amigos e colegas interessados em Python e vamos crescer juntos na nossa comunidade de programadores!

InfinitePy Newsletter - Sua fonte para aprendizado e inspiração em Python.