Python e MongoDB com PyMongo — 2025 | Python Brasil

Aprenda a usar MongoDB com Python usando PyMongo. Tutorial completo com CRUD, queries avancadas, indices e agregacoes para aplicacoes reais.

5 min de leitura Equipe Python Brasil

MongoDB e o banco de dados NoSQL mais popular do mercado, e Python e uma das linguagens que melhor se integra com ele. Usando PyMongo, a biblioteca oficial, voce consegue fazer operacoes de CRUD, queries complexas e agregacoes de forma intuitiva. Neste guia, a gente vai montar tudo do zero com exemplos praticos.

Instalacao e Conexao

Primeiro, instale o PyMongo via pip:

pip install pymongo

Para conectar ao MongoDB, use o MongoClient:

from pymongo import MongoClient

# Conexao local
client = MongoClient("mongodb://localhost:27017/")

# Ou com MongoDB Atlas (nuvem)
# client = MongoClient("mongodb+srv://usuario:senha@cluster.mongodb.net/")

# Selecionar banco de dados
db = client["minha_aplicacao"]

# Selecionar colecao (equivalente a tabela)
usuarios = db["usuarios"]

print("Conexao estabelecida com sucesso!")
print(f"Bancos disponiveis: {client.list_database_names()}")

Operacoes CRUD Basicas

Inserir Documentos

from datetime import datetime

# Inserir um documento
novo_usuario = {
    "nome": "Ana Silva",
    "email": "ana@email.com",
    "idade": 28,
    "linguagens": ["python", "javascript"],
    "ativo": True,
    "criado_em": datetime.now()
}

resultado = usuarios.insert_one(novo_usuario)
print(f"ID inserido: {resultado.inserted_id}")

# Inserir varios documentos de uma vez
novos_usuarios = [
    {
        "nome": "Bruno Costa",
        "email": "bruno@email.com",
        "idade": 32,
        "linguagens": ["python", "go"],
        "ativo": True,
        "criado_em": datetime.now()
    },
    {
        "nome": "Carla Souza",
        "email": "carla@email.com",
        "idade": 25,
        "linguagens": ["python", "rust"],
        "ativo": False,
        "criado_em": datetime.now()
    },
    {
        "nome": "Diego Lima",
        "email": "diego@email.com",
        "idade": 30,
        "linguagens": ["python", "java"],
        "ativo": True,
        "criado_em": datetime.now()
    }
]

resultado = usuarios.insert_many(novos_usuarios)
print(f"IDs inseridos: {resultado.inserted_ids}")

Consultar Documentos

# Buscar um documento
usuario = usuarios.find_one({"email": "ana@email.com"})
print(f"Encontrado: {usuario['nome']}")

# Buscar todos os documentos
for u in usuarios.find():
    print(f"{u['nome']} - {u['email']}")

# Buscar com filtro
ativos = usuarios.find({"ativo": True})
for u in ativos:
    print(f"Ativo: {u['nome']}")

# Selecionar apenas campos especificos (projecao)
nomes = usuarios.find(
    {"ativo": True},
    {"nome": 1, "email": 1, "_id": 0}
)
for u in nomes:
    print(u)
# {'nome': 'Ana Silva', 'email': 'ana@email.com'}

Atualizar Documentos

# Atualizar um documento
resultado = usuarios.update_one(
    {"email": "carla@email.com"},
    {"$set": {"ativo": True, "idade": 26}}
)
print(f"Modificados: {resultado.modified_count}")

# Atualizar varios documentos
resultado = usuarios.update_many(
    {"ativo": True},
    {"$set": {"ultimo_acesso": datetime.now()}}
)
print(f"Modificados: {resultado.modified_count}")

# Adicionar item a um array
usuarios.update_one(
    {"email": "ana@email.com"},
    {"$push": {"linguagens": "typescript"}}
)

# Incrementar valor numerico
usuarios.update_one(
    {"email": "ana@email.com"},
    {"$inc": {"idade": 1}}
)

Deletar Documentos

# Deletar um documento
resultado = usuarios.delete_one({"email": "diego@email.com"})
print(f"Deletados: {resultado.deleted_count}")

# Deletar varios documentos
resultado = usuarios.delete_many({"ativo": False})
print(f"Deletados: {resultado.deleted_count}")

Queries Avancadas

O MongoDB oferece operadores poderosos para consultas complexas:

# Operadores de comparacao
# Usuarios com idade maior que 25
jovens = usuarios.find({"idade": {"$gt": 25}})

# Idade entre 20 e 30
faixa = usuarios.find({"idade": {"$gte": 20, "$lte": 30}})

# Operador $in - valor em uma lista
python_ou_go = usuarios.find(
    {"linguagens": {"$in": ["python", "go"]}}
)

# Operador $regex - busca por padrao
ana = usuarios.find(
    {"nome": {"$regex": "^Ana", "$options": "i"}}
)

# Operadores logicos
# $and, $or, $not
resultado = usuarios.find({
    "$or": [
        {"idade": {"$lt": 25}},
        {"linguagens": "rust"}
    ]
})

# Ordenacao e limite
top_usuarios = usuarios.find().sort("idade", -1).limit(5)
for u in top_usuarios:
    print(f"{u['nome']}: {u['idade']} anos")

# Contar documentos
total_ativos = usuarios.count_documents({"ativo": True})
print(f"Total de usuarios ativos: {total_ativos}")

Indices para Performance

Indices sao essenciais para manter consultas rapidas conforme a base cresce:

# Criar indice simples
usuarios.create_index("email", unique=True)

# Criar indice composto
usuarios.create_index([("nome", 1), ("idade", -1)])

# Indice de texto para busca full-text
usuarios.create_index([("nome", "text")])
resultado = usuarios.find({"$text": {"$search": "Ana Silva"}})

# Listar indices existentes
for indice in usuarios.list_indexes():
    print(indice)

# Verificar plano de execucao de uma query
plano = usuarios.find({"email": "ana@email.com"}).explain()
print(f"Estrategia: {plano['queryPlanner']['winningPlan']}")

Agregacoes

O pipeline de agregacao e uma das funcionalidades mais poderosas do MongoDB:

# Pipeline de agregacao
pipeline = [
    # Filtrar apenas ativos
    {"$match": {"ativo": True}},

    # Agrupar por faixa etaria
    {"$group": {
        "_id": {
            "$switch": {
                "branches": [
                    {"case": {"$lt": ["$idade", 25]}, "then": "jovem"},
                    {"case": {"$lt": ["$idade", 35]}, "then": "adulto"},
                ],
                "default": "senior"
            }
        },
        "total": {"$sum": 1},
        "media_idade": {"$avg": "$idade"}
    }},

    # Ordenar por total
    {"$sort": {"total": -1}}
]

resultados = usuarios.aggregate(pipeline)
for r in resultados:
    print(f"Faixa: {r['_id']}, Total: {r['total']}, Media: {r['media_idade']:.1f}")

# Agregacao com $unwind para arrays
pipeline_linguagens = [
    {"$unwind": "$linguagens"},
    {"$group": {
        "_id": "$linguagens",
        "total": {"$sum": 1}
    }},
    {"$sort": {"total": -1}},
    {"$limit": 5}
]

top_linguagens = usuarios.aggregate(pipeline_linguagens)
for lang in top_linguagens:
    print(f"{lang['_id']}: {lang['total']} usuarios")

Classe de Repositorio Completa

Para projetos reais, encapsule as operacoes em uma classe:

from pymongo import MongoClient, ASCENDING
from datetime import datetime
from typing import Optional
from bson import ObjectId


class UsuarioRepository:
    """Repositorio para operacoes com usuarios no MongoDB."""

    def __init__(self, connection_string: str, database: str):
        self.client = MongoClient(connection_string)
        self.db = self.client[database]
        self.collection = self.db["usuarios"]
        self._criar_indices()

    def _criar_indices(self):
        self.collection.create_index("email", unique=True)
        self.collection.create_index([("nome", ASCENDING)])

    def criar(self, nome: str, email: str, idade: int) -> str:
        documento = {
            "nome": nome,
            "email": email,
            "idade": idade,
            "ativo": True,
            "criado_em": datetime.now()
        }
        resultado = self.collection.insert_one(documento)
        return str(resultado.inserted_id)

    def buscar_por_id(self, id: str) -> Optional[dict]:
        return self.collection.find_one({"_id": ObjectId(id)})

    def buscar_por_email(self, email: str) -> Optional[dict]:
        return self.collection.find_one({"email": email})

    def listar(self, filtro: dict = None, limite: int = 100) -> list:
        return list(self.collection.find(filtro or {}).limit(limite))

    def atualizar(self, id: str, dados: dict) -> bool:
        resultado = self.collection.update_one(
            {"_id": ObjectId(id)},
            {"$set": dados}
        )
        return resultado.modified_count > 0

    def deletar(self, id: str) -> bool:
        resultado = self.collection.delete_one({"_id": ObjectId(id)})
        return resultado.deleted_count > 0


# Uso
repo = UsuarioRepository("mongodb://localhost:27017/", "minha_app")
user_id = repo.criar("Fernanda", "fernanda@email.com", 29)
usuario = repo.buscar_por_id(user_id)
print(usuario)

Boas Praticas

Ao trabalhar com MongoDB e Python, lembre-se dessas recomendacoes:

  • Sempre crie indices para campos usados em consultas frequentes
  • Use insert_many ao inves de multiplos insert_one para insercoes em lote
  • Limite o tamanho dos documentos, o maximo e 16 MB
  • Use projecoes para retornar apenas os campos necessarios
  • Trate conexoes com try/except e feche o client quando nao precisar mais
  • Valide os dados antes de inserir no banco usando bibliotecas como Pydantic

Conclusao

MongoDB com PyMongo e uma combinacao poderosa para aplicacoes que precisam de flexibilidade no schema e performance em leituras. Com as operacoes CRUD, queries avancadas, indices e agregacoes apresentadas neste guia, voce tem uma base solida para construir aplicacoes robustas. Comece com o modelo de repositorio e adapte conforme as necessidades do seu projeto.

E

Equipe Python Brasil

Contribuidor do Python Brasil — Aprenda Python em Português