Conectando seu site a um banco de dados MongoDB

Por que MongoDB para projetos Flask

O MongoDB é um banco de dados NoSQL orientado a documentos. Em vez de tabelas e linhas, você trabalha com coleções e documentos JSON. Isso combina muito bem com o Flask por alguns motivos:

Flexibilidade de esquema — você não precisa definir colunas antes de inserir dados. Isso acelera a prototipagem e permite evoluir a estrutura dos dados sem migrações.
Formato nativo — os documentos são armazenados em BSON (JSON binário), o que facilita a comunicação entre o banco e a API, já que o Flask trabalha naturalmente com dicionários Python.
Escalabilidade horizontal — o MongoDB foi projetado para distribuir dados entre múltiplos servidores, o que é útil quando sua aplicação cresce.

Se o seu projeto tem dados com estrutura variável ou você quer começar rápido sem se preocupar com esquemas rígidos, o MongoDB é uma ótima escolha.

Instalação do PyMongo e conexão

O PyMongo é o driver oficial do MongoDB para Python. Instale junto com o Flask:

pip install flask pymongo

Para conectar, basta informar a URI do MongoDB. Em desenvolvimento local, o padrão é localhost na porta 27017:

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["meu_banco"]

A variável db agora representa o banco de dados. A partir dela, você acessa qualquer coleção.

Estrutura: onde colocar a conexão no app

Em projetos pequenos, você pode criar a conexão diretamente no arquivo principal. Porém, à medida que o projeto cresce, é melhor isolar a configuração do banco em um módulo separado.

Uma abordagem comum é criar um arquivo database.py:

# database.py
import os
from pymongo import MongoClient

MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017/")
client = MongoClient(MONGO_URI)
db = client["flask_app"]

Depois, no seu arquivo principal ou nas rotas, basta importar:

# app.py
from flask import Flask
from database import db

app = Flask(__name__)
tarefas = db["tarefas"]

Dessa forma, toda a aplicação compartilha a mesma conexão, e você pode trocar a URI via variável de ambiente em produção sem alterar o código.

Create: insert_one e insert_many

Para inserir um único documento, use insert_one. Ele retorna o _id gerado automaticamente pelo MongoDB:

resultado = tarefas.insert_one({
    "titulo": "Estudar PyMongo",
    "concluida": False
})
print(resultado.inserted_id)

Para inserir vários documentos de uma vez, use insert_many:

novas_tarefas = [
    {"titulo": "Configurar MongoDB", "concluida": True},
    {"titulo": "Criar rotas CRUD", "concluida": False},
    {"titulo": "Escrever testes", "concluida": False},
]
resultado = tarefas.insert_many(novas_tarefas)
print(resultado.inserted_ids)

Read: find_one, find com filtros e projeções

Para buscar um único documento, use find_one. Ele retorna o primeiro documento que corresponde ao filtro ou None se não encontrar nada:

tarefa = tarefas.find_one({"titulo": "Estudar PyMongo"})
print(tarefa)

Para buscar vários documentos, use find. Ele retorna um cursor que você pode iterar:

pendentes = tarefas.find({"concluida": False})
for tarefa in pendentes:
    print(tarefa["titulo"])

Você também pode usar projeções para controlar quais campos retornam na consulta. Isso é útil para evitar tráfego desnecessário:

# Retorna apenas o titulo, sem o _id
resultados = tarefas.find(
    {"concluida": False},
    {"titulo": 1, "_id": 0}
)

Para buscar por _id, lembre-se de converter a string para ObjectId:

from bson.objectid import ObjectId

tarefa = tarefas.find_one({"_id": ObjectId("664f1a2b3c4d5e6f7a8b9c0d")})

Update: update_one com $set

O operador $set atualiza apenas os campos especificados, sem sobrescrever o documento inteiro:

tarefas.update_one(
    {"titulo": "Estudar PyMongo"},
    {"$set": {"concluida": True}}
)

O update_one retorna um objeto com informações sobre a operação. Você pode verificar quantos documentos foram modificados:

resultado = tarefas.update_one(
    {"titulo": "Estudar PyMongo"},
    {"$set": {"concluida": True}}
)
print(resultado.modified_count)  # 1

Delete: delete_one

Para remover um documento, use delete_one com um filtro:

resultado = tarefas.delete_one({"titulo": "Estudar PyMongo"})
print(resultado.deleted_count)  # 1

Cuidado: se o filtro corresponder a mais de um documento, apenas o primeiro será removido. Para remover vários, use delete_many.

Índices e unique constraints

Índices melhoram a velocidade das consultas. Sem eles, o MongoDB precisa percorrer todos os documentos da coleção para encontrar o que você busca.

Para criar um índice simples:

tarefas.create_index("titulo")

Para garantir que um campo seja único (por exemplo, impedir tarefas duplicadas):

tarefas.create_index("titulo", unique=True)

Com o unique=True, qualquer tentativa de inserir um documento com um título que já existe vai gerar um erro DuplicateKeyError. Você pode tratar isso no código:

from pymongo.errors import DuplicateKeyError

try:
    tarefas.insert_one({"titulo": "Estudar PyMongo", "concluida": False})
except DuplicateKeyError:
    print("Tarefa com esse titulo ja existe.")

Exemplo prático: CRUD completo com rotas Flask

Agora vamos juntar tudo em uma aplicação Flask funcional. Este exemplo implementa um CRUD completo para uma coleção de tarefas:

from flask import Flask, request, jsonify
from pymongo import MongoClient
from pymongo.errors import DuplicateKeyError
from bson.objectid import ObjectId

app = Flask(__name__)

client = MongoClient("mongodb://localhost:27017/")
db = client["flask_app"]
tarefas = db["tarefas"]
tarefas.create_index("titulo", unique=True)


def serializar(doc):
    doc["_id"] = str(doc["_id"])
    return doc


@app.route("/tarefas", methods=["POST"])
def criar_tarefa():
    dados = request.get_json()
    try:
        resultado = tarefas.insert_one({
            "titulo": dados["titulo"],
            "concluida": False
        })
    except DuplicateKeyError:
        return jsonify({"erro": "Tarefa ja existe"}), 409
    return jsonify({"id": str(resultado.inserted_id)}), 201


@app.route("/tarefas", methods=["GET"])
def listar_tarefas():
    filtro = {}
    status = request.args.get("concluida")
    if status is not None:
        filtro["concluida"] = status.lower() == "true"
    docs = tarefas.find(filtro)
    return jsonify([serializar(doc) for doc in docs])


@app.route("/tarefas/<id>", methods=["GET"])
def buscar_tarefa(id):
    doc = tarefas.find_one({"_id": ObjectId(id)})
    if not doc:
        return jsonify({"erro": "Tarefa nao encontrada"}), 404
    return jsonify(serializar(doc))


@app.route("/tarefas/<id>", methods=["PUT"])
def atualizar_tarefa(id):
    dados = request.get_json()
    campos = {}
    if "titulo" in dados:
        campos["titulo"] = dados["titulo"]
    if "concluida" in dados:
        campos["concluida"] = dados["concluida"]
    resultado = tarefas.update_one(
        {"_id": ObjectId(id)},
        {"$set": campos}
    )
    if resultado.matched_count == 0:
        return jsonify({"erro": "Tarefa nao encontrada"}), 404
    return jsonify({"modificados": resultado.modified_count})


@app.route("/tarefas/<id>", methods=["DELETE"])
def deletar_tarefa(id):
    resultado = tarefas.delete_one({"_id": ObjectId(id)})
    if resultado.deleted_count == 0:
        return jsonify({"erro": "Tarefa nao encontrada"}), 404
    return jsonify({"removidos": resultado.deleted_count})

Para testar, rode o servidor com flask run e use curl ou qualquer cliente HTTP:

# Criar tarefa
curl -X POST http://localhost:5000/tarefas \
  -H "Content-Type: application/json" \
  -d '{"titulo": "Aprender MongoDB"}'

# Listar todas
curl http://localhost:5000/tarefas

# Filtrar por status
curl http://localhost:5000/tarefas?concluida=false

# Atualizar
curl -X PUT http://localhost:5000/tarefas/<ID> \
  -H "Content-Type: application/json" \
  -d '{"concluida": true}'

# Deletar
curl -X DELETE http://localhost:5000/tarefas/<ID>

Com esse exemplo, você tem uma API REST completa usando Flask e MongoDB. A partir daqui, você pode adicionar validação de dados, paginação nas listagens e autenticação nas rotas.