Python e PostgreSQL: Guia Completo
Conecte Python ao PostgreSQL usando psycopg2 e SQLAlchemy. Aprenda CRUD, queries avancadas, migrations e boas praticas de acesso a banco.
PostgreSQL e um dos bancos de dados relacionais mais robustos e populares do mundo, e Python oferece ferramentas excelentes para trabalhar com ele. Neste guia, vamos desde a conexao basica com psycopg2 ate o uso avancado do SQLAlchemy como ORM completo.
Configurando o Ambiente
Antes de comecar, instale as dependencias necessarias:
pip install psycopg2-binary sqlalchemy alembic
O pacote psycopg2-binary e a versao pre-compilada do driver PostgreSQL para Python. Para producao, recomenda-se usar psycopg2 com as bibliotecas do sistema instaladas.
Certifique-se de ter um PostgreSQL rodando. Voce pode usar Docker para isso:
docker run --name pg-dev -e POSTGRES_PASSWORD=senha123 \
-e POSTGRES_DB=meuprojeto -p 5432:5432 -d postgres:16
Conexao Direta com Psycopg2
O psycopg2 e o driver mais utilizado para conectar Python ao PostgreSQL. Ele oferece controle total sobre as queries SQL:
import psycopg2
from psycopg2.extras import RealDictCursor
def conectar():
"""Cria conexao com o PostgreSQL."""
return psycopg2.connect(
host="localhost",
port=5432,
database="meuprojeto",
user="postgres",
password="senha123",
)
# Criando tabela
def criar_tabela():
conn = conectar()
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS produtos (
id SERIAL PRIMARY KEY,
nome VARCHAR(200) NOT NULL,
preco NUMERIC(10, 2) NOT NULL,
estoque INTEGER DEFAULT 0,
criado_em TIMESTAMP DEFAULT NOW()
)
""")
conn.commit()
cursor.close()
conn.close()
criar_tabela()
Operacoes CRUD com Psycopg2
Vamos implementar as quatro operacoes basicas de forma segura, usando parametros para evitar SQL injection:
def inserir_produto(nome: str, preco: float, estoque: int) -> int:
"""Insere um produto e retorna o ID gerado."""
conn = conectar()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO produtos (nome, preco, estoque) VALUES (%s, %s, %s) RETURNING id",
(nome, preco, estoque),
)
produto_id = cursor.fetchone()[0]
conn.commit()
cursor.close()
conn.close()
return produto_id
def listar_produtos() -> list[dict]:
"""Lista todos os produtos como dicionarios."""
conn = conectar()
cursor = conn.cursor(cursor_factory=RealDictCursor)
cursor.execute("SELECT * FROM produtos ORDER BY nome")
produtos = cursor.fetchall()
cursor.close()
conn.close()
return [dict(p) for p in produtos]
def atualizar_preco(produto_id: int, novo_preco: float) -> bool:
"""Atualiza o preco de um produto."""
conn = conectar()
cursor = conn.cursor()
cursor.execute(
"UPDATE produtos SET preco = %s WHERE id = %s",
(novo_preco, produto_id),
)
atualizado = cursor.rowcount > 0
conn.commit()
cursor.close()
conn.close()
return atualizado
def deletar_produto(produto_id: int) -> bool:
"""Remove um produto pelo ID."""
conn = conectar()
cursor = conn.cursor()
cursor.execute("DELETE FROM produtos WHERE id = %s", (produto_id,))
deletado = cursor.rowcount > 0
conn.commit()
cursor.close()
conn.close()
return deletado
# Uso
novo_id = inserir_produto("Notebook Dell", 4599.90, 25)
print(f"Produto criado com ID: {novo_id}")
produtos = listar_produtos()
for p in produtos:
print(f" {p['nome']}: R$ {p['preco']}")
Gerenciamento de Conexoes com Context Manager
Abrir e fechar conexoes manualmente e propenso a erros. Use context managers para garantir que os recursos sejam liberados:
from contextlib import contextmanager
@contextmanager
def get_conexao():
"""Context manager para conexao com o banco."""
conn = psycopg2.connect(
host="localhost",
database="meuprojeto",
user="postgres",
password="senha123",
)
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
@contextmanager
def get_cursor(conn):
"""Context manager para cursor."""
cursor = conn.cursor(cursor_factory=RealDictCursor)
try:
yield cursor
finally:
cursor.close()
# Uso limpo e seguro
with get_conexao() as conn:
with get_cursor(conn) as cur:
cur.execute("SELECT * FROM produtos WHERE preco > %s", (1000,))
caros = cur.fetchall()
for p in caros:
print(f"{p['nome']}: R$ {p['preco']}")
Usando SQLAlchemy como ORM
O SQLAlchemy e o ORM mais poderoso do ecossistema Python. Ele abstrai o SQL e permite trabalhar com objetos Python:
from sqlalchemy import create_engine, Column, Integer, String, Numeric, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime
DATABASE_URL = "postgresql://postgres:senha123@localhost:5432/meuprojeto"
engine = create_engine(DATABASE_URL, echo=False)
Session = sessionmaker(bind=engine)
Base = declarative_base()
class Produto(Base):
__tablename__ = "produtos_orm"
id = Column(Integer, primary_key=True, autoincrement=True)
nome = Column(String(200), nullable=False)
preco = Column(Numeric(10, 2), nullable=False)
estoque = Column(Integer, default=0)
criado_em = Column(DateTime, default=datetime.utcnow)
def __repr__(self):
return f"<Produto(id={self.id}, nome='{self.nome}', preco={self.preco})>"
def to_dict(self):
return {
"id": self.id,
"nome": self.nome,
"preco": float(self.preco),
"estoque": self.estoque,
}
# Criar tabela
Base.metadata.create_all(engine)
CRUD com SQLAlchemy
As operacoes com o ORM sao mais intuitivas e seguras:
def criar_produto_orm(nome: str, preco: float, estoque: int) -> Produto:
session = Session()
produto = Produto(nome=nome, preco=preco, estoque=estoque)
session.add(produto)
session.commit()
session.refresh(produto)
session.close()
return produto
def buscar_produtos_por_faixa(preco_min: float, preco_max: float) -> list[dict]:
session = Session()
produtos = (
session.query(Produto)
.filter(Produto.preco.between(preco_min, preco_max))
.order_by(Produto.preco)
.all()
)
resultado = [p.to_dict() for p in produtos]
session.close()
return resultado
def atualizar_estoque(produto_id: int, quantidade: int) -> bool:
session = Session()
produto = session.query(Produto).filter_by(id=produto_id).first()
if produto:
produto.estoque += quantidade
session.commit()
session.close()
return True
session.close()
return False
# Exemplo de uso
notebook = criar_produto_orm("Monitor LG 27'", 1899.90, 15)
print(f"Criado: {notebook}")
faixa = buscar_produtos_por_faixa(1000, 3000)
for p in faixa:
print(f" {p['nome']}: R$ {p['preco']:.2f}")
Migrations com Alembic
Para gerenciar alteracoes no schema do banco, use o Alembic junto com o SQLAlchemy:
alembic init migrations
Configure o alembic.ini com a URL do banco e crie migracoes automaticamente:
alembic revision --autogenerate -m "criar tabela produtos"
alembic upgrade head
Isso permite versionar o schema do banco de dados junto com o codigo, facilitando deploys e trabalho em equipe.
Boas Praticas
Ao trabalhar com Python e PostgreSQL, siga algumas praticas fundamentais. Sempre use queries parametrizadas para evitar SQL injection. Nunca concatene strings para montar SQL. Utilize connection pooling em aplicacoes web com SQLAlchemy ou psycopg2.pool. Mantenha as credenciais em variaveis de ambiente, nunca no codigo-fonte. Crie indices para colunas usadas em filtros e ordenacoes frequentes. E sempre feche conexoes e cursores apos o uso.
import os
# Credenciais via variavel de ambiente
DATABASE_URL = os.environ.get(
"DATABASE_URL",
"postgresql://postgres:senha123@localhost:5432/meuprojeto"
)
Conclusao
Python e PostgreSQL formam uma dupla extremamente produtiva. O psycopg2 oferece controle total sobre o SQL, enquanto o SQLAlchemy eleva a produtividade com um ORM completo. Combine ambos com Alembic para migrations e voce tera uma stack de dados robusta e profissional. Comece pelo psycopg2 para entender os fundamentos e migre para o SQLAlchemy conforme seu projeto crescer.
Equipe Python Brasil
Contribuidor do Python Brasil — Aprenda Python em Português