MySQL, MariaDB, PostgreSQL e MongoDB Já leu

No artigo anterior 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.")

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)

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']}")

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

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.utcnow(),
        "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.utcnow()
                }
            }
        }
    )
    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', [])}")

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"])

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 anterior 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")

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

Resumo

• MySQL e MariaDB são compatíveis em API — use mysql-connector-python ou pymysql
• PostgreSQL oferece recursos avançados: arrays nativos, JSONB, RETURNING, e transações ACID completas
• execute_values do psycopg2 realiza inserções em lote de forma eficiente
• MongoDB armazena documentos BSON sem esquema fixo — ideal para dados variáveis
• O Aggregation Pipeline do MongoDB é a ferramenta central para análises complexas
• SQLAlchemy conecta a MySQL e PostgreSQL apenas trocando a connection string
• Sempre use variáveis de ambiente para credenciais — nunca escreva senhas no código

Referências e Leituras Complementares

• 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.