MySQL, MariaDB, PostgreSQL e MongoDB

No artigo Banco de Dados com SQLite e SQLAlchemy trabalhamos com SQLite — excelente para desenvolvimento local. Em produção, a maioria dos sistemas utiliza bancos mais robustos: MySQL e MariaDB para aplicações web tradicionais, PostgreSQL para sistemas que exigem recursos avançados, e MongoDB quando os dados têm estrutura variável ou documento-orientada. Neste artigo veremos como conectar e operar cada um deles a partir do Python.

MySQL e MariaDB com mysql-connector-python

MySQL e MariaDB são compatíveis em termos de API — o mesmo driver funciona para ambos.

pip install mysql-connector-python

Conectando

import mysql.connector
from mysql.connector import Error

def criar_conexao(host, usuario, senha, banco=None):
    try:
        conn = mysql.connector.connect(
            host=host,
            user=usuario,
            password=senha,
            database=banco,
            charset="utf8mb4",
            collation="utf8mb4_unicode_ci"
        )
        if conn.is_connected():
            print(f"Conectado ao MySQL — versão: {conn.get_server_info()}")
        return conn
    except Error as e:
        print(f"Erro ao conectar: {e}")
        return None


conn = criar_conexao("localhost", "root", "senha123", "escola")

Criando banco e tabelas

def configurar_banco(conn):
    cursor = conn.cursor()


    cursor.execute("CREATE DATABASE IF NOT EXISTS escola CHARACTER SET utf8mb4")
    cursor.execute("USE escola")

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS alunos (
            id        INT AUTO_INCREMENT PRIMARY KEY,
            nome      VARCHAR(100) NOT NULL,
            email     VARCHAR(100) UNIQUE NOT NULL,
            nota      DECIMAL(4,2) DEFAULT 0.00,
            ativo     TINYINT(1)   DEFAULT 1,
            criado_em TIMESTAMP    DEFAULT CURRENT_TIMESTAMP
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS disciplinas (
            id   INT AUTO_INCREMENT PRIMARY KEY,
            nome VARCHAR(80) NOT NULL
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS notas (
            aluno_id      INT,
            disciplina_id INT,
            valor         DECIMAL(4,2),
            PRIMARY KEY (aluno_id, disciplina_id),
            FOREIGN KEY (aluno_id)      REFERENCES alunos(id),
            FOREIGN KEY (disciplina_id) REFERENCES disciplinas(id)
        ) ENGINE=InnoDB
    """)

    conn.commit()
    print("Banco configurado.")

Dois cuidados com esse começo. O configurar_banco cria o banco escola, mas o criar_conexao do bloco anterior já se conecta a ele, e numa instalação nova isso falha com Unknown database 'escola': para a primeira configuração, conecte com banco=None. E o criar_conexao engole o erro e devolve None, de modo que a falha real aparece uma linha depois, como AttributeError: 'NoneType' object has no attribute 'cursor', longe da causa. Em código de produção, deixe a exceção de conexão subir.

CRUD

def inserir_aluno(conn, nome, email, nota):
    cursor = conn.cursor()
    sql    = "INSERT INTO alunos (nome, email, nota) VALUES (%s, %s, %s)"
    cursor.execute(sql, (nome, email, nota))
    conn.commit()
    return cursor.lastrowid   # ID gerado


def buscar_alunos(conn, nota_minima=0.0):
    cursor = conn.cursor(dictionary=True)   # retorna dicts
    sql    = """
        SELECT id, nome, email, nota
        FROM   alunos
        WHERE  nota >= %s AND ativo = 1
        ORDER  BY nota DESC
    """
    cursor.execute(sql, (nota_minima,))
    return cursor.fetchall()


def atualizar_nota(conn, aluno_id, nova_nota):
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE alunos SET nota = %s WHERE id = %s",
        (nova_nota, aluno_id)
    )
    conn.commit()
    return cursor.rowcount   # linhas afetadas


# Uso
conn = criar_conexao("localhost", "root", "senha123", "escola")

if conn:
    id1 = inserir_aluno(conn, "Ana Silva",   "ana@email.com",   9.5)
    id2 = inserir_aluno(conn, "Bruno Costa", "bruno@email.com", 7.0)
    id3 = inserir_aluno(conn, "Carla Souza", "carla@email.com", 5.5)

    print("\nAlunos com nota >= 7.0:")
    for aluno in buscar_alunos(conn, 7.0):
        print(f"  [{aluno['id']}] {aluno['nome']:15} — {aluno['nota']}")

    conn.close()

Usando PyMySQL como alternativa

pip install pymysql

import pymysql

conn = pymysql.connect(
    host="localhost",
    user="root",
    password="senha123",
    database="escola",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor
)

with conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM alunos WHERE nota >= %s", (7.0,))
        for row in cursor.fetchall():
            print(row)

Os valores de coluna DECIMAL chegam ao Python como decimal.Decimal, e não como float — é o que preserva a precisão, e é também por isso que a nota aparece como 9.50. A consequência prática é que aluno["nota"] * 1.1 levanta TypeError: unsupported operand type(s) for *: 'decimal.Decimal' and 'float'; a conta se faz com Decimal("1.1"), ou com inteiros. O PostgreSQL faz o mesmo com NUMERIC.

Outro detalhe do PyMySQL que engana quem vem de outro driver: o with conn: dele fecha a conexão, mas não faz commit. Como o autocommit vem desligado, um INSERT feito dentro do bloco sem conn.commit() explícito é descartado ao fechar, sem erro. O mysql-connector-python se comporta do mesmo jeito. E, com o MySQL 8, cujo método de autenticação padrão é caching_sha2_password, o PyMySQL pode precisar do extra de criptografia: pip install "PyMySQL[rsa]".

PostgreSQL com psycopg2

PostgreSQL é o banco relacional de código aberto mais avançado — suporta JSON nativo, arrays, full-text search, transações robustas e muito mais.

pip install psycopg2-binary

Conectando e criando tabelas

import psycopg2
from psycopg2.extras import RealDictCursor, execute_values

def criar_conexao_pg():
    return psycopg2.connect(
        host="localhost",
        port=5432,
        dbname="escola",
        user="postgres",
        password="senha123"
    )


def configurar_banco_pg():
    with criar_conexao_pg() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS alunos (
                    id        SERIAL PRIMARY KEY,
                    nome      VARCHAR(100) NOT NULL,
                    email     VARCHAR(100) UNIQUE NOT NULL,
                    nota      NUMERIC(4,2) DEFAULT 0.0,
                    ativo     BOOLEAN      DEFAULT TRUE,
                    tags      TEXT[],
                    metadata  JSONB,
                    criado_em TIMESTAMPTZ  DEFAULT NOW()
                )
            """)
            conn.commit()
            print("Banco PostgreSQL configurado.")

Recursos exclusivos do PostgreSQL

import json

def inserir_com_metadados(nome, email, nota, tags, metadata):
    with criar_conexao_pg() as conn:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute("""
                INSERT INTO alunos (nome, email, nota, tags, metadata)
                VALUES (%s, %s, %s, %s, %s)
                RETURNING id, nome, criado_em
            """, (
                nome, email, nota,
                tags,                         # array nativo
                json.dumps(metadata)          # JSONB
            ))
            conn.commit()
            return dict(cur.fetchone())


def buscar_por_tag(tag: str):
    """Busca usando operador de array do PostgreSQL."""
    with criar_conexao_pg() as conn:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute("""
                SELECT id, nome, nota, tags
                FROM   alunos
                WHERE  %s = ANY(tags)
                ORDER  BY nota DESC
            """, (tag,))
            return cur.fetchall()


def buscar_por_metadata(chave: str, valor: str):
    """Busca dentro de campo JSONB."""
    with criar_conexao_pg() as conn:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute("""
                SELECT id, nome, metadata
                FROM   alunos
                WHERE  metadata ->> %s = %s
            """, (chave, valor))
            return cur.fetchall()


def inserir_em_lote(alunos: list):
    """Inserção em lote com execute_values — muito mais rápido."""
    with criar_conexao_pg() as conn:
        with conn.cursor() as cur:
            execute_values(cur, """
                INSERT INTO alunos (nome, email, nota)
                VALUES %s
                ON CONFLICT (email) DO UPDATE SET nota = EXCLUDED.nota
            """, [(a["nome"], a["email"], a["nota"]) for a in alunos])
            conn.commit()


# Uso
resultado = inserir_com_metadados(
    nome="Ana Silva",
    email="ana@email.com",
    nota=9.5,
    tags=["monitor", "destaque", "bolsista"],
    metadata={"cidade": "Recife", "turno": "manhã"}
)
print(f"Inserido: {resultado}")

alunos_lote = [
    {"nome": "Bruno Costa", "email": "bruno@email.com", "nota": 7.0},
    {"nome": "Carla Souza", "email": "carla@email.com", "nota": 8.5},
    {"nome": "Diego Lima",  "email": "diego@email.com", "nota": 5.5},
]
inserir_em_lote(alunos_lote)

print("\nAlunos com tag 'monitor':")
for a in buscar_por_tag("monitor"):
    print(f"  {a['nome']} — {a['nota']}")

Aqui o with tem um terceiro significado. No psycopg2, with conn: delimita uma transação — faz commit na saída normal e rollback na exceção — e não fecha a conexão: medido, depois do bloco, conn.closed continua 0. Nas funções acima quem fecha é o coletor de lixo, quando a função retorna e a última referência some; basta guardar a conexão num atributo ou numa lista para ela ficar aberta até o servidor recusar novas conexões. Para fechar de forma garantida no psycopg2, use contextlib.closing(psycopg2.connect(...)) por fora do with da transação. No psycopg 3, a seguir, o with faz as duas coisas.

psycopg3 — a versão moderna

pip install psycopg[binary]

import psycopg

with psycopg.connect("host=localhost dbname=escola user=postgres password=senha123") as conn:
    with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
        cur.execute("SELECT * FROM alunos WHERE nota >= %s", (7.0,))
        for row in cur.fetchall():
            print(row)

MongoDB com pymongo

MongoDB é um banco de dados orientado a documentos — armazena dados em formato BSON (similar a JSON), sem esquema fixo. Ideal quando os dados têm estrutura variável ou evoluem rapidamente.

pip install pymongo

Conectando

from pymongo import MongoClient
from pymongo.errors import ConnectionFailure
from datetime import datetime, UTC

def criar_cliente():
    cliente = MongoClient(
        "mongodb://localhost:27017/",
        serverSelectionTimeoutMS=5000
    )
    try:
        cliente.admin.command("ping")
        print("Conectado ao MongoDB.")
    except ConnectionFailure:
        print("Falha na conexão com MongoDB.")
    return cliente


cliente = criar_cliente()
db      = cliente["escola"]
alunos  = db["alunos"]        # collection — cria automaticamente

CRUD no MongoDB

# CREATE — inserindo documentos
def inserir_aluno_mongo(nome, email, nota, tags=None):
    documento = {
        "nome":      nome,
        "email":     email,
        "nota":      nota,
        "ativo":     True,
        "tags":      tags or [],
        "criado_em": datetime.now(UTC),
        "historico": []
    }
    resultado = alunos.insert_one(documento)
    return str(resultado.inserted_id)


def inserir_varios(lista):
    resultado = alunos.insert_many(lista)
    return [str(id) for id in resultado.inserted_ids]


# READ — consultando documentos
def buscar_aprovados(nota_minima=6.0):
    cursor = alunos.find(
        {"nota": {"$gte": nota_minima}, "ativo": True},
        {"_id": 0, "nome": 1, "nota": 1, "tags": 1}  # projeção
    ).sort("nota", -1)
    return list(cursor)


def buscar_por_tag(tag: str):
    return list(alunos.find(
        {"tags": tag},
        {"nome": 1, "nota": 1, "tags": 1}
    ))


# UPDATE — atualizando documentos
def atualizar_nota_mongo(email, nova_nota):
    resultado = alunos.update_one(
        {"email": email},
        {
            "$set": {"nota": nova_nota},
            "$push": {
                "historico": {
                    "nota":  nova_nota,
                    "data":  datetime.now(UTC)
                }
            }
        }
    )
    return resultado.modified_count


def adicionar_tag(email, tag):
    alunos.update_one(
        {"email": email},
        {"$addToSet": {"tags": tag}}   # addToSet evita duplicatas
    )


# DELETE — removendo documentos
def desativar_aluno_mongo(email):
    alunos.update_one(
        {"email": email},
        {"$set": {"ativo": False}}
    )


# Uso
inserir_aluno_mongo("Ana Silva",   "ana@email.com",   9.5, ["monitor", "bolsista"])
inserir_aluno_mongo("Bruno Costa", "bruno@email.com", 7.0)
inserir_aluno_mongo("Carla Souza", "carla@email.com", 8.5, ["destaque"])

atualizar_nota_mongo("bruno@email.com", 7.5)
adicionar_tag("bruno@email.com", "monitor")

print("\nAprovados:")
for a in buscar_aprovados():
    print(f"  {a['nome']:15} — {a['nota']} {a.get('tags', [])}")

Sobre as datas: o BSON guarda o instante em UTC com precisão de milissegundos, e o pymongo, por padrão, devolve-o como datetime naive. Um valor gravado como 14h30 de São Paulo volta como 17:30 sem fuso, e os microssegundos são truncados (.123456 volta .123000). Para receber datas aware, crie o cliente com MongoClient(..., tz_aware=True). E como o MongoDB não impõe esquema, ele também não impõe tipo: um documento com "nota": "9.5", em texto, é aceito, e depois simplesmente não aparece num filtro {"nota": {"$gte": 6}}, porque o $gte numérico não compara com strings.

Aggregation Pipeline

O recurso mais poderoso do MongoDB para análises:

def estatisticas_mongo():
    pipeline = [
        {"$match": {"ativo": True}},
        {"$group": {
            "_id":       None,
            "total":     {"$sum": 1},
            "media":     {"$avg": "$nota"},
            "maior":     {"$max": "$nota"},
            "menor":     {"$min": "$nota"},
            "aprovados": {
                "$sum": {
                    "$cond": [{"$gte": ["$nota", 6.0]}, 1, 0]
                }
            }
        }}
    ]
    resultado = list(alunos.aggregate(pipeline))
    return resultado[0] if resultado else {}


def distribuicao_por_conceito():
    pipeline = [
        {"$match": {"ativo": True}},
        {"$project": {
            "nome": 1,
            "nota": 1,
            "conceito": {
                "$switch": {
                    "branches": [
                        {"case": {"$gte": ["$nota", 9]}, "then": "Excelente"},
                        {"case": {"$gte": ["$nota", 7]}, "then": "Bom"},
                        {"case": {"$gte": ["$nota", 6]}, "then": "Regular"},
                    ],
                    "default": "Insuficiente"
                }
            }
        }},
        {"$group": {
            "_id":       "$conceito",
            "quantidade": {"$sum": 1},
            "alunos":    {"$push": "$nome"}
        }},
        {"$sort": {"quantidade": -1}}
    ]
    return list(alunos.aggregate(pipeline))


stats = estatisticas_mongo()
print(f"\nEstatísticas MongoDB:")
print(f"  Total:     {stats.get('total')}")
print(f"  Média:     {stats.get('media', 0):.2f}")
print(f"  Aprovados: {stats.get('aprovados')}")

print("\nDistribuição por conceito:")
for grupo in distribuicao_por_conceito():
    print(f"  {grupo['_id']:15}: {grupo['quantidade']} aluno(s)")

Índices no MongoDB

from pymongo import ASCENDING, DESCENDING, TEXT

# Índice simples
alunos.create_index("email", unique=True)

# Índice composto
alunos.create_index([("nota", DESCENDING), ("ativo", ASCENDING)])

# Índice de texto para busca full-text
alunos.create_index([("nome", TEXT)])

# Buscando com índice de texto
resultados = list(alunos.find({"$text": {"$search": "Silva"}}))

# Listando índices existentes
for indice in alunos.list_indexes():
    print(indice["name"])

O índice único em email deveria vir antes dos dados, não depois. Sem ele, execute o bloco de CRUD duas vezes e a coleção aceita os mesmos três alunos de novo, sem erro. Criado depois, sobre dados já duplicados, o create_index(..., unique=True) falha com DuplicateKeyError. Com o índice no lugar, a segunda execução para no primeiro insert_one, com E11000 duplicate key error — que é o comportamento desejado.

Comparativo: Quando Usar Cada Banco

Usando com SQLAlchemy

SQLAlchemy funciona com MySQL e PostgreSQL — basta trocar a connection string:

from sqlalchemy import create_engine

# MySQL
engine_mysql = create_engine(
    "mysql+pymysql://usuario:senha@localhost:3306/escola",
    pool_size=5,
    max_overflow=10
)

# MariaDB
engine_maria = create_engine(
    "mariadb+pymysql://usuario:senha@localhost:3306/escola"
)

# PostgreSQL
engine_pg = create_engine(
    "postgresql+psycopg2://usuario:senha@localhost:5432/escola",
    pool_size=5,
    max_overflow=10,
    echo=False
)

Os modelos ORM do artigo Banco de Dados com SQLite e SQLAlchemy funcionam sem alteração — apenas a engine muda.

Variáveis de Ambiente para Credenciais

Nunca escreva senhas diretamente no código:

pip install python-dotenv

# arquivo .env
DB_HOST=localhost
DB_PORT=5432
DB_NAME=escola
DB_USER=postgres
DB_PASSWORD=senha_super_secreta
MONGO_URI=mongodb://localhost:27017/

from dotenv import load_dotenv
import os

load_dotenv()

DB_URL = (
    f"postgresql+psycopg2://{os.getenv('DB_USER')}:"
    f"{os.getenv('DB_PASSWORD')}@"
    f"{os.getenv('DB_HOST')}:"
    f"{os.getenv('DB_PORT')}/"
    f"{os.getenv('DB_NAME')}"
)

MONGO_URI = os.getenv("MONGO_URI")

Montar a URL com f-string tem um defeito que só aparece com a senha de produção: caracteres como @, :, / e # têm significado dentro de uma URL. Com a senha p@ss:w/rd#1, o SQLAlchemy lê o w como porta e levanta ValueError: invalid literal for int() with base 10: 'w'. O URL.create monta a URL a partir das partes e escapa cada uma:

from sqlalchemy import URL, create_engine

url = URL.create(
    "postgresql+psycopg2",
    username=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),   # escapada automaticamente
    host=os.getenv("DB_HOST"),
    port=int(os.getenv("DB_PORT", "5432")),
    database=os.getenv("DB_NAME"),
)
engine = create_engine(url)

Adicione .env ao .gitignore — nunca suba credenciais para o repositório.

A partir do Python, os três bancos relacionais se usam quase do mesmo jeito: conexão, cursor, consulta com parâmetros e commit. As diferenças que custam caro estão nos detalhes que parecem iguais. O marcador de parâmetro é %s no MySQL e no PostgreSQL, e não ? como no SQLite. DECIMAL e NUMERIC chegam como Decimal, que não se mistura com float. E o with conn: muda de sentido conforme o driver: no psycopg2 é transação sem fechamento, no PyMySQL é fechamento sem commit, e só no psycopg 3 é as duas coisas. O PostgreSQL acrescenta o que os outros não têm na mesma medida — arrays, JSONB indexável, RETURNING e ON CONFLICT —, e o SQLAlchemy permite que o mesmo modelo rode em qualquer um deles trocando a URL.

O MongoDB troca a tabela pelo documento, e com isso troca também quem garante a consistência. O banco aceita qualquer formato, então a regra passa a morar no código e no que se declara explicitamente: o índice único antes dos dados, um validador $jsonSchema quando o tipo importa, tz_aware=True para as datas não voltarem sem fuso, e um replica set quando for preciso transação. O Aggregation Pipeline faz no servidor o que em SQL seria GROUP BY. E, em qualquer dos quatro, a credencial vem do ambiente, nunca do código, e a URL de conexão se monta com URL.create, que escapa a senha.

Fontes e leituras recomendadas

• mysql-connector-python — https://dev.mysql.com/doc/connector-python/en/
• psycopg2 — documentação oficial — https://www.psycopg.org/docs/
• psycopg3 — versão moderna — https://www.psycopg.org/psycopg3/docs/
• pymongo — documentação oficial — https://pymongo.readthedocs.io/en/stable/
• MongoDB Aggregation Pipeline — https://www.mongodb.com/docs/manual/aggregation/
• python-dotenv — https://pypi.org/project/python-dotenv/
• CHODOROW, Kristina. MongoDB: The Definitive Guide. 3. ed. O'Reilly Media, 2019. — referência completa sobre MongoDB.
• MOMJIAN, Bruce. PostgreSQL: Introduction and Concepts. Addison-Wesley, 2001. — fundamentos do PostgreSQL por um dos seus principais desenvolvedores.

Exercícios

Exercício 1

Uma equipe migra um serviço de PostgreSQL para MySQL. O código usava psycopg2 no padrão with conn: with conn.cursor() as cur: cur.execute("INSERT ..."), sem nenhum commit() explícito, e funcionava. A troca foi para PyMySQL, mudando só a função de conexão. Os testes passam, não há exceção nenhuma, e em produção nenhum cadastro novo aparece no banco. Explique e diga como escrever o código para que ele se comporte igual nos dois drivers.

Exercício 2

Uma rotina sincroniza cadastros de uma planilha para o PostgreSQL com execute_values e ON CONFLICT (email) DO UPDATE SET nota = EXCLUDED.nota, em lotes de mil. Num dia, o lote inteiro falha com CardinalityViolation: ON CONFLICT DO UPDATE command cannot affect row a second time, e nenhum dos mil registros é gravado. Os dados parecem normais. O que pode haver na planilha, e como tornar a rotina robusta?

Exercício 3

Uma API Flask com SQLAlchemy e PostgreSQL funciona o dia inteiro. Toda manhã, a primeira requisição depois da madrugada falha com OperationalError: server closed the connection unexpectedly, e as seguintes funcionam. A equipe de infraestrutura informa que um firewall derruba conexões ociosas depois de uma hora. Explique por que só a primeira falha e como resolver no código.

Exercício 4

A coluna metadata, em JSONB, passou a guardar também o ano de ingresso, como número: {"cidade": "Recife", "ano": 2024}. Um desenvolvedor reaproveita buscar_por_metadata("ano", 2024) deste artigo e recebe UndefinedFunction: operator does not exist: text = integer. Com "2024", entre aspas, funciona. Explique o erro, avalie o contorno e proponha uma consulta melhor.

Exercício 5

Um relatório de aprovados feito sobre o MongoDB, com find({"nota": {"$gte": 6}}), mostra 1.240 alunos, e a secretaria sabe que são mais. Descobre-se que parte dos documentos veio de um importador antigo que gravava a nota como texto, "9.5". Não houve erro em momento nenhum. Explique por que esses alunos somem, como encontrá-los e como impedir que aconteça de novo.