Noções Básicas de Bancos de Dados

9 exemplos para fundamentos de banco de dados em Python - 6 básicos e 3 intermediários.

Pré-requisitos

uv pip install "sqlalchemy>=2.0" alembic asyncpg psycopg[binary]

• O estilo SQLAlchemy 2.0 é assumido em toda esta seção.

Exemplos Básicos

1. Conexão DB-API

Conecte-se com um driver PEP 249.

import sqlite3
 
with sqlite3.connect("app.db") as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS items (id INTEGER PRIMARY KEY, name TEXT)")
    conn.execute("INSERT INTO items(name) VALUES (?)", ("widget",))

• O gerenciador de contexto faz commit em caso de sucesso.
• Consultas parametrizadas previnem injeção de SQL.
• SQLite apenas para desenvolvimento local em escala.

Relacionado: SQLAlchemy Core - camada de engine

2. Transações

Faça commit e rollback explicitamente.

conn.execute("BEGIN")
try:
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except Exception:
    conn.rollback()
    raise

• Mantenha as transações curtas.
• Uma unidade lógica por transação.
• Trate exceções com rollback.

Relacionado: Gerenciamento de Conexão - escopo do pool

3. SQLAlchemy Engine

Crie um engine e execute texto.

from sqlalchemy import create_engine, text
engine = create_engine("sqlite:///app.db")
with engine.begin() as conn:
    conn.execute(text("SELECT 1"))

• O engine gerencia conexões.
• begin() faz commit automaticamente.
• Use strings de URL de dialeto consistentemente.

4. Modelo Declarativo ORM

Mapeie uma classe para uma tabela.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
class Base(DeclarativeBase):
    pass
 
class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()

• Mapped[] anota as colunas.
• DeclarativeBase é SQLAlchemy 2.0.
• Crie tabelas via metadados ou Alembic.

Relacionado: SQLAlchemy ORM 2.0 - sessões

5. Consulta de Sessão

Leia linhas com uma sessão.

from sqlalchemy import select
from sqlalchemy.orm import Session
 
with Session(engine) as session:
    rows = session.scalars(select(Item)).all()

• A sessão rastreia o estado do objeto.
• select() substitui a API de consulta legada.
• O gerenciador de contexto fecha a sessão.

6. Inserir Linha ORM

Persista uma nova entidade.

with Session(engine) as session:
    session.add(Item(name="bolt"))
    session.commit()

• add() anexa à sessão.
• commit() envia o SQL.
• refresh() recarrega chaves geradas.

Exemplos Intermediários

7. Alembic Init

Inicie o histórico de migração.

# alembic init alembic
# edite alembic.ini sqlalchemy.url
# alembic revision --autogenerate -m "init"
# alembic upgrade head

• autogenerate compara modelos com o banco de dados.
• Revise as migrações manualmente.
• Faça upgrade no pipeline de implantação.

Relacionado: Migrações com Alembic - fluxos de trabalho

8. Engine Assíncrono

Acesso não bloqueante ao PostgreSQL.

from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

• Driver asyncpg para Postgres.
• Emparelhe com rotas FastAPI assíncronas.
• pool_pre_ping evita sockets obsoletos.

Relacionado: Acesso Assíncrono a Banco de Dados - sessões

9. Cache Redis

Armazene em cache leituras frequentes.

import redis
r = redis.Redis.from_url("redis://localhost:6379/0")
r.setex("item:1", 60, "cached-payload")

• Chaves com TTL expiram automaticamente.
• Padrão "cache aside" em serviços.
• Trate o cache como opcional - a fonte da verdade permanece o banco de dados.

Relacionado: NoSQL e Cache - padrões

Versões das Pilhas: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.