<h2>Introdução ao MongoDB e PyMongo</h2>
<p>MongoDB é um banco de dados NoSQL orientado a documentos que armazena dados em formato JSON (ou BSON — Binary JSON). Diferentemente de bancos relacionais tradicionais, MongoDB oferece flexibilidade estrutural, permitindo que documentos na mesma coleção tenham esquemas diferentes. Python, através da biblioteca <strong>PyMongo</strong>, fornece um driver oficial para interagir com o MongoDB de forma simples e intuitiva.</p>
<p>PyMongo é a biblioteca padrão para conectar aplicações Python ao MongoDB. Ela abstrai a complexidade do protocolo de comunicação e oferece uma API Pythônica para realizar operações CRUD (Create, Read, Update, Delete), gerenciar índices e executar consultas complexas. Antes de avançarmos para operações assíncronas e agregações, é fundamental compreender como estabelecer conexões básicas e manipular documentos.</p>
<h3>Instalação e Configuração Básica</h3>
<p>Para começar, instale o PyMongo via pip:</p>
<pre><code class="language-bash">pip install pymongo</code></pre>
<p>Agora, vamos criar uma conexão simples com o MongoDB:</p>
<pre><code class="language-python">from pymongo import MongoClient
from datetime import datetime
Conectar ao servidor MongoDB local
client = MongoClient('mongodb://localhost:27017/')
Acessar um banco de dados
db = client['minha_aplicacao']
Acessar uma coleção
usuarios = db['usuarios']
Inserir um documento
documento = {
'nome': 'João Silva',
'email': 'joao@example.com',
'idade': 28,
'criado_em': datetime.now()
}
resultado = usuarios.insert_one(documento)
print(f"ID inserido: {resultado.inserted_id}")</code></pre>
<p>Neste exemplo, criamos uma conexão com o MongoDB, acessamos um banco chamado <code>minha_aplicacao</code> e a coleção <code>usuarios</code>. A operação <code>insert_one()</code> insere um único documento e retorna um objeto com o ID gerado automaticamente pelo MongoDB.</p>
<h3>Operações CRUD Essenciais</h3>
<p>As operações CRUD são fundamentais. Vamos explorar cada uma:</p>
<pre><code class="language-python">from pymongo import MongoClient
from datetime import datetime
client = MongoClient('mongodb://localhost:27017/')
db = client['minha_aplicacao']
usuarios = db['usuarios']
CREATE - Inserir múltiplos documentos
novos_usuarios = [
{'nome': 'Maria', 'email': 'maria@example.com', 'idade': 25},
{'nome': 'Pedro', 'email': 'pedro@example.com', 'idade': 30},
{'nome': 'Ana', 'email': 'ana@example.com', 'idade': 27}
]
resultado = usuarios.insert_many(novos_usuarios)
print(f"IDs inseridos: {resultado.inserted_ids}")
READ - Buscar um documento
usuario = usuarios.find_one({'nome': 'Maria'})
print(f"Usuário encontrado: {usuario['email']}")
READ - Buscar múltiplos documentos
usuarios_maiores_26 = usuarios.find({'idade': {'$gte': 26}})
for user in usuarios_maiores_26:
print(f"{user['nome']} tem {user['idade']} anos")
UPDATE - Atualizar um documento
usuarios.update_one(
{'nome': 'Maria'},
{'$set': {'idade': 26, 'atualizado_em': datetime.now()}}
)
UPDATE - Atualizar múltiplos documentos
usuarios.update_many(
{'idade': {'$lt': 28}},
{'$inc': {'idade': 1}} # Incrementa a idade em 1
)
DELETE - Deletar um documento
usuarios.delete_one({'nome': 'Pedro'})
DELETE - Deletar múltiplos documentos
usuarios.delete_many({'idade': {'$lt': 25}})</code></pre>
<p>Compreender esses operadores MongoDB (<code>$gte</code>, <code>$set</code>, <code>$inc</code>) é crucial. Eles representam a linguagem de consulta do MongoDB e permitem operações complexas de forma declarativa.</p>
<h2>Motor: Programação Assíncrona com MongoDB</h2>
<p>Quando sua aplicação precisa lidar com múltiplas conexões simultâneas (servidores web, APIs), as operações síncronas no PyMongo podem se tornar gargalos. <strong>Motor</strong> é uma biblioteca que torna o PyMongo totalmente assíncrono, permitindo que sua aplicação não bloqueie durante operações de banco de dados.</p>
<p>Motor funciona como um wrapper assíncrono do PyMongo, mantendo a mesma API, mas retornando corrotinas que devem ser aguardadas com <code>await</code>. Isso é particularmente útil em frameworks como FastAPI, Quart e aiohttp, que também são assíncronos por natureza.</p>
<h3>Instalação e Conceitos Fundamentais</h3>
<p>Instale o Motor:</p>
<pre><code class="language-bash">pip install motor</code></pre>
<p>Antes de escrever código, entenda que o Motor utiliza <code>async/await</code>, o padrão de programação assíncrona do Python moderno. Quando você aguarda uma operação assíncrona, a thread é liberada para processar outras requisições, maximizando a utilização de recursos.</p>
<pre><code class="language-python">import asyncio
from motor.motor_asyncio import AsyncClient
from datetime import datetime
async def main():
Conectar ao MongoDB usando Motor
client = AsyncClient('mongodb://localhost:27017/')
db = client['minha_aplicacao']
usuarios = db['usuarios']
Inserir um documento de forma assíncrona
resultado = await usuarios.insert_one({
'nome': 'Lucas',
'email': 'lucas@example.com',
'idade': 24
})
print(f"Documento inserido com ID: {resultado.inserted_id}")
Buscar um documento
usuario = await usuarios.find_one({'nome': 'Lucas'})
print(f"Usuário: {usuario}")
Buscar múltiplos documentos de forma assíncrona
cursor = usuarios.find({'idade': {'$gte': 20}})
async for user in cursor:
print(f"{user['nome']}: {user['idade']} anos")
Fechar a conexão
client.close()
Executar a corrotina
asyncio.run(main())</code></pre>
<p>A diferença fundamental é que cada operação retorna uma corrotina que deve ser aguardada com <code>await</code>. Isso permite que o event loop do Python processe outras tarefas enquanto o banco de dados processa a requisição.</p>
<h3>Integrando Motor com FastAPI</h3>
<p>Motor brilha quando integrado com frameworks web assíncronos. Aqui está um exemplo com FastAPI:</p>
<pre><code class="language-python">from fastapi import FastAPI, HTTPException
from motor.motor_asyncio import AsyncClient
from pydantic import BaseModel
from typing import Optional
from bson import ObjectId
app = FastAPI()
Configuração do banco de dados
MONGODB_URL = "mongodb://localhost:27017"
client = AsyncClient(MONGODB_URL)
db = client['minha_aplicacao']
Fechar a conexão ao desligar a aplicação
@app.on_event("shutdown")
async def shutdown_db():
client.close()
Modelo Pydantic para validação
class Usuario(BaseModel):
nome: str
email: str
idade: int
class UsuarioResponse(Usuario):
id: str
class Config:
from_attributes = True
@app.post("/usuarios/")
async def criar_usuario(usuario: Usuario):
collection = db['usuarios']
resultado = await collection.insert_one(usuario.dict())
return {"id": str(resultado.inserted_id), **usuario.dict()}
@app.get("/usuarios/{usuario_id}")
async def obter_usuario(usuario_id: str):
collection = db['usuarios']
try:
documento = await collection.find_one({"_id": ObjectId(usuario_id)})
if not documento:
raise HTTPException(status_code=404, detail="Usuário não encontrado")
documento['id'] = str(documento['_id'])
return documento
except Exception as e:
raise HTTPException(status_code=400, detail=f"ID inválido: {str(e)}")
@app.get("/usuarios/")
async def listar_usuarios(skip: int = 0, limit: int = 10):
collection = db['usuarios']
cursor = collection.find().skip(skip).limit(limit)
usuarios = []
async for doc in cursor:
doc['id'] = str(doc['_id'])
usuarios.append(doc)
return usuarios
@app.put("/usuarios/{usuario_id}")
async def atualizar_usuario(usuario_id: str, usuario: Usuario):
collection = db['usuarios']
try:
resultado = await collection.update_one(
{"_id": ObjectId(usuario_id)},
{"$set": usuario.dict()}
)
if resultado.matched_count == 0:
raise HTTPException(status_code=404, detail="Usuário não encontrado")
return {"mensagem": "Usuário atualizado com sucesso"}
except Exception as e:
raise HTTPException(status_code=400, detail=f"Erro: {str(e)}")
@app.delete("/usuarios/{usuario_id}")
async def deletar_usuario(usuario_id: str):
collection = db['usuarios']
try:
resultado = await collection.delete_one({"_id": ObjectId(usuario_id)})
if resultado.deleted_count == 0:
raise HTTPException(status_code=404, detail="Usuário não encontrado")
return {"mensagem": "Usuário deletado com sucesso"}
except Exception as e:
raise HTTPException(status_code=400, detail=f"Erro: {str(e)}")</code></pre>
<p>Este exemplo demonstra como Motor integra perfeitamente com FastAPI, mantendo a aplicação responsiva mesmo sob carga elevada. Cada requisição HTTP é processada de forma assíncrona, e enquanto aguarda a resposta do banco de dados, o event loop pode processar outras requisições.</p>
<h2>Agregação: Processamento Avançado de Dados</h2>
<p>A agregação no MongoDB é um framework poderoso para transformar, filtrar e processar documentos no próprio servidor. Diferentemente de buscas simples, agregações permitem operações complexas como agrupamento, projeção, ordenação e cálculos, tudo de forma eficiente no lado do servidor. Isso reduz drasticamente a quantidade de dados transferidos para a aplicação.</p>
<h3>Conceitos de Pipeline de Agregação</h3>
<p>Uma agregação no MongoDB funciona como um pipeline: documentos passam por uma série de estágios (<code>stages</code>), cada um transformando os dados até obter o resultado final. Os estágios mais comuns são:</p>
<ul>
<li><strong>$match</strong>: Filtra documentos (equivalente a WHERE no SQL)</li>
<li><strong>$group</strong>: Agrupa documentos por um campo específico</li>
<li><strong>$project</strong>: Seleciona/renomeia campos</li>
<li><strong>$sort</strong>: Ordena documentos</li>
<li><strong>$limit</strong>: Limita o número de documentos</li>
<li><strong>$lookup</strong>: Join com outra coleção</li>
<li><strong>$unwind</strong>: Desconstrói arrays em documentos separados</li>
<li><strong>$addFields</strong>: Adiciona novos campos calculados</li>
</ul>
<p>Vamos ilustrar com um exemplo prático:</p>
<pre><code class="language-python">from pymongo import MongoClient
from datetime import datetime, timedelta
client = MongoClient('mongodb://localhost:27017/')
db = client['loja']
vendas = db['vendas']
Inserir dados de exemplo
vendas.insert_many([
{
'data': datetime.now() - timedelta(days=10),
'vendedor': 'João',
'produto': 'Notebook',
'quantidade': 2,
'preco_unitario': 3000
},
{
'data': datetime.now() - timedelta(days=8),
'vendedor': 'Maria',
'produto': 'Mouse',
'quantidade': 10,
'preco_unitario': 50
},
{
'data': datetime.now() - timedelta(days=5),
'vendedor': 'João',
'produto': 'Teclado',
'quantidade': 5,
'preco_unitario': 150
},
{
'data': datetime.now() - timedelta(days=3),
'vendedor': 'Maria',
'produto': 'Monitor',
'quantidade': 1,
'preco_unitario': 1200
},
])
Pipeline de agregação: Total de vendas por vendedor
pipeline = [
{
'$group': {
'_id': '$vendedor',
'total_vendido': {
'$sum': {'$multiply': ['$quantidade', '$preco_unitario']}
},
'quantidade_produtos': {'$sum': '$quantidade'}
}
},
{
'$sort': {'total_vendido': -1}
}
]
resultado = list(vendas.aggregate(pipeline))
for doc in resultado:
print(f"Vendedor: {doc['_id']}, Total: R$ {doc['total_vendido']}, Quantidade: {doc['quantidade_produtos']}")</code></pre>
<p>Este pipeline primeiro agrupa as vendas por vendedor, calculando o total de vendas e a quantidade total de produtos, depois ordena pelo total de vendas em ordem decrescente.</p>
<h3>Agregações Complexas com Lookups e Unwinding</h3>
<p>Para casos mais avançados, combinamos múltiplos estágios:</p>
<pre><code class="language-python">from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
db = client['escola']
Criar coleções de exemplo
alunos = db['alunos']
notas = db['notas']
Limpar dados anteriores
alunos.delete_many({})
notas.delete_many({})
Inserir alunos
alunos.insert_many([
{'_id': 1, 'nome': 'Alice'},
{'_id': 2, 'nome': 'Bob'},
{'_id': 3, 'nome': 'Carlos'}
])
Inserir notas (referenciando alunos)
notas.insert_many([
{'aluno_id': 1, 'materia': 'Matemática', 'nota': 8.5},
{'aluno_id': 1, 'materia': 'Português', 'nota': 9.0},
{'aluno_id': 2, 'materia': 'Matemática', 'nota': 7.0},
{'aluno_id': 2, 'materia': 'Português', 'nota': 8.0},
{'aluno_id': 3, 'materia': 'Matemática', 'nota': 9.5},
{'aluno_id': 3, 'materia': 'Português', 'nota': 7.5},
])
Pipeline: Média de notas por aluno
pipeline = [
{
'$group': {
'_id': '$aluno_id',
'media': {'$avg': '$nota'},
'notas_obtidas': {'$push': '$nota'}
}
},
{
'$lookup': {
'from': 'alunos',
'localField': '_id',
'foreignField': '_id',
'as': 'info_aluno'
}
},
{
'$unwind': '$info_aluno'
},
{
'$project': {
'_id': 0,
'nome': '$info_aluno.nome',
'media': {'$round': ['$media', 2]},
'total_notas': {'$size': '$notas_obtidas'},
'situacao': {
'$cond': [
{'$gte': ['$media', 7]},
'Aprovado',
'Reprovado'
]
}
}
},
{
'$sort': {'media': -1}
}
]
resultado = list(notas.aggregate(pipeline))
for doc in resultado:
print(f"{doc['nome']}: Média {doc['media']} - {doc['situacao']} ({doc['total_notas']} notas)")</code></pre>
<p>Neste exemplo complexo, fazemos:</p>
<ol>
<li>Agrupamos notas por aluno, calculando a média</li>
<li>Fazemos um <code>lookup</code> (join) com a coleção de alunos</li>
<li>Desconstruímos o array de alunos com <code>$unwind</code></li>
<li>Projetamos campos customizados com lógica condicional (<code>$cond</code>)</li>
<li>Ordenamos pelo resultado final</li>
</ol>
<h3>Agregação Assíncrona com Motor</h3>
<p>Motor também oferece suporte a agregações assíncronas:</p>
<pre><code class="language-python">import asyncio
from motor.motor_asyncio import AsyncClient
async def agregacao_assincrona():
client = AsyncClient('mongodb://localhost:27017/')
db = client['vendas_online']
pedidos = db['pedidos']
Inserir dados de exemplo
await pedidos.insert_many([
{'cliente': 'Alice', 'valor': 150.00, 'status': 'entregue'},
{'cliente': 'Bob', 'valor': 320.00, 'status': 'entregue'},
{'cliente': 'Alice', 'valor': 80.00, 'status': 'pendente'},
{'cliente': 'Carlos', 'valor': 450.00, 'status': 'entregue'},
{'cliente': 'Bob', 'valor': 200.00, 'status': 'cancelado'},
])
Pipeline de agregação
pipeline = [
{
'$match': {'status': 'entregue'}
},
{
'$group': {
'_id': '$cliente',
'total_gasto': {'$sum': '$valor'},
'quantidade_pedidos': {'$sum': 1}
}
},
{
'$sort': {'total_gasto': -1}
}
]
Executar agregação de forma assíncrona
cursor = pedidos.aggregate(pipeline)
print("=== Clientes com mais compras entregues ===")
async for doc in cursor:
print(f"{doc['_id']}: R$ {doc['total_gasto']:.2f} ({doc['quantidade_pedidos']} pedidos)")
client.close()
asyncio.run(agregacao_assincrona())</code></pre>
<p>Este exemplo demonstra como usar agregações com Motor, permitindo processamento eficiente de grandes volumes de dados sem bloquear a aplicação.</p>
<h3>Dicas de Performance em Agregações</h3>
<p>Agregações são poderosas, mas devem ser otimizadas. Use índices nos campos usados em <code>$match</code> para melhorar drasticamente a performance. Posicione <code>$match</code> o mais cedo possível no pipeline, reduzindo o volume de dados processados nos estágios subsequentes. Evite <code>$lookup</code> em coleções muito grandes, pois isso tem custo computacional elevado.</p>
<pre><code class="language-python"># Exemplo de criação de índices
vendas = db['vendas']
vendas.create_index('vendedor') # Índice simples
vendas.create_index([('data', -1), ('vendedor', 1)]) # Índice composto</code></pre>
<h2>Conclusão</h2>
<p>Nesta aula aprofundada, você aprendeu que <strong>MongoDB com Python oferece flexibilidade estrutural aliada a simplicidade de uso</strong>, seja através do PyMongo para aplicações síncronas ou Motor para sistemas de alta concorrência. A escolha entre PyMongo e Motor depende da arquitetura de sua aplicação: se você está construindo APIs assíncronas modernas com FastAPI, Quart ou aiohttp, Motor é a solução natural; caso contrário, PyMongo oferece simplicidade sem comprometer funcionalidade.</p>
<p>Em segundo lugar, <strong>agregações são o coração do processamento avançado no MongoDB</strong>, permitindo que você realize operações complexas no servidor em vez de trazer dados brutos para a aplicação. Dominar pipelines de agregação transforma-o de um usuário básico a um profissional capaz de extrair insights de dados em tempo real, otimizando performance e reduzindo consumo de banda.</p>
<p>Por fim, <strong>a combinação de Motor + FastAPI + Agregações cria aplicações escaláveis e eficientes</strong>, capazes de servir centenas de requisições simultâneas sem degradação de performance. Pratique construindo pequenos projetos, domine os operadores de agregação, e você terá domínio completo dessa stack moderna e poderosa.</p>
<h2>Referências</h2>
<ul>
<li><a href="https://pymongo.readthedocs.io/en/stable/" target="_blank" rel="noopener noreferrer">Documentação Oficial do PyMongo</a></li>
<li><a href="https://motor.readthedocs.io/" target="_blank" rel="noopener noreferrer">Documentação do Motor - Async Driver para MongoDB</a></li>
<li><a href="https://docs.mongodb.com/manual/reference/operator/aggregation/" target="_blank" rel="noopener noreferrer">MongoDB Aggregation Pipeline Documentation</a></li>
<li><a href="https://fastapi.tiangolo.com/tutorial/sql-databases/" target="_blank" rel="noopener noreferrer">FastAPI com Banco de Dados</a></li>
<li><a href="https://realpython.com/motor-motor-asyncio-mongodb-python/" target="_blank" rel="noopener noreferrer">Real Python - Motor and Asyncio</a></li>
</ul>
<p><!-- FIM --></p>