Db Migration

Database Migration Skill — GesComIA

Guia para criar migrations de banco de dados seguindo os padrões do projeto GesComIA.

Stack: PostgreSQL + SQLAlchemy + Alembic

---

Regras de Padronização (OBRIGATÓRIAS)

1. Nomenclatura

2. Tipos de Dados

2.1 Identificação do Cliente (OBRIGATÓRIO)

Toda tabela que pertence a um cliente/empresa DEVE ter:

-- Identificação do cliente (OBRIGATÓRIO para tabelas tenant)
cliente_id uuid REFERENCES cliente(id) ON DELETE CASCADE NOT NULL,

-- Campo de compatibilidade (código antigo integer)
cod_cliente int4,

No modelo SQLAlchemy:

cliente_id = Column(UUID(as_uuid=True), ForeignKey("cliente.id", ondelete="CASCADE"), nullable=False)
cod_cliente = Column(Integer)  # Mantido para compatibilidade

No Token JWT (sempre incluir):

payload = {
    "sub": str(usuario.id),
    "cod_cliente": usuario.cod_cliente,  # int (compatibilidade)
    "cliente_id": str(usuario.cliente_id),  # UUID
    "cli_cnpj": cliente.cnpj,
    ...
}

No UserInfo (sempre incluir):

class UserInfo(BaseModel):
    cod_cliente_usuario: str  # UUID
    usu_usuario: str
    cod_cliente: int | None  # int (compatibilidade)
    cliente_id: str | None  # UUID
    cli_cnpj: str
    cli_razao: str

3. Campos Obrigatórios (Auditoria)

Toda tabela DEVE ter:

-- Auditoria
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
data_criacao timestamp with time zone DEFAULT now(),
data_atualizacao timestamp with time zone DEFAULT now(),
deletado boolean DEFAULT false

4. Constraints Obrigatórias

-- Unique
cnpj varchar(18) UNIQUE NOT NULL,

-- Index para busca frequente
CREATE INDEX idx_usuario_usuario ON usuario(usuario);
CREATE INDEX idx_usuario_email ON usuario(email);

5. Segurança

---

Fase 1: Análise da Tabela Atual

Checklist de análise

Para cada tabela, responder:

1. PK atual: serial → migrar para uuid 2. Prefixos a remover: cli_, usu_, etc. 3. Campos sensíveis: senha, senha_hash, dados pessoais 4. Unique constraints: quais campos? 5. Indexes necessários: para busca/login? 6. Foreign Keys: referenciar outras tabelas (usar UUID) 7. Campos de auditoria: faltando? 8. Booleanos: usar bpchar(1) → boolean

---

Fase 2: Gerar Script SQL

Template de Migration

-- Migration: cliente
-- Autor: [nome]
-- Data: [YYYY-MM-DD]
-- Descrição: Padroniza tabela cliente para novos padrões

BEGIN;

-- 1. Criar nova tabela com padrões
CREATE TABLE public.cliente_new (
    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    sincronizado boolean DEFAULT false,
    data_hora timestamp with time zone DEFAULT now(),
    deletado boolean DEFAULT false,
    
    -- Dados principais
    razao varchar(60) NOT NULL,
    fantasia varchar(60),
    cnpj varchar(18) UNIQUE NOT NULL,
    insc_est varchar(18),
    insc_mun varchar(18),
    cnae varchar(7),
    
    -- Endereço
    cep varchar(9),
    logradouro varchar(60),
    numero varchar(10),
    bairro varchar(60),
    cidade varchar(60),
    complemento varchar(60),
    ibge varchar(7),
    uf varchar(2) DEFAULT 'BR',
    
    -- Contato
    telefone varchar(15),
    celular varchar(15),
    whatsapp varchar(15),
    email varchar(60),
    contato varchar(60),
    site varchar(60),
    
    -- Redes sociais
    facebook varchar(60),
    twitter varchar(60),
    instagram varchar(60),
    youtube varchar(60),
    
    -- Metadata
    somente_leitura boolean DEFAULT false,
    data_criacao timestamp with time zone DEFAULT now(),
    data_atualizacao timestamp with time zone DEFAULT now()
);

-- 2. Migrar dados (mapeando colunas antigas)
INSERT INTO cliente_new (id, razao, fantasia, cnpj, insc_est, insc_mun, cnae, 
    cep, logradouro, numero, bairro, cidade, complemento, ibge, uf,
    telefone, celular, whatsapp, email, contato, site,
    facebook, twitter, instagram, youtube, somente_leitura, data_criacao)
SELECT 
    gen_random_uuid(),
    cli_razao,
    cli_fantasia,
    cli_cnpj,
    cli_insc_est,
    cli_insc_mun,
    cli_cnae,
    cli_cep,
    cli_logradouro,
    cli_numero,
    cli_bairro,
    cli_cidade,
    cli_complemento,
    cli_ibge,
    cli_uf,
    cli_telefone,
    cli_celular,
    cli_whatsapp,
    cli_email,
    cli_contato,
    cli_site,
    cli_facebook,
    cli_twitter,
    cli_instagram,
    cli_youtube,
    cli_somente_leitura = 'S',
    COALESCE(cli_data_hora, now())
FROM cliente;

-- 3. Dropar tabela antiga
DROP TABLE IF EXISTS cliente CASCADE;

-- 4. Renomear nova tabela
ALTER TABLE cliente_new RENAME TO cliente;

-- 5. Recriar sequences (se necessário)
CREATE SEQUENCE IF NOT EXISTS cliente_cod_cliente_seq;

COMMIT;

---

Fase 3: Atualizar Modelos SQLAlchemy

Template de Modelo

# app/models/central/cliente.py
from uuid import uuid4
from sqlalchemy import Column, String, Boolean, DateTime
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship

from app.db.central import CentralBase


class Cliente(CentralBase):
    __tablename__ = "cliente"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    razao = Column(String(60), nullable=False)
    fantasia = Column(String(60))
    cnpj = Column(String(18), unique=True, nullable=False, index=True)
    insc_est = Column(String(18))
    insc_mun = Column(String(18))
    cnae = Column(String(7))
    
    # Endereço
    cep = Column(String(9))
    logradouro = Column(String(60))
    numero = Column(String(10))
    bairro = Column(String(60))
    cidade = Column(String(60))
    complemento = Column(String(60))
    ibge = Column(String(7))
    uf = Column(String(2), default="BR")
    
    # Contato
    telefone = Column(String(15))
    celular = Column(String(15))
    whatsapp = Column(String(15))
    email = Column(String(60))
    contato = Column(String(60))
    site = Column(String(60))
    
    # Sociais
    facebook = Column(String(60))
    twitter = Column(String(60))
    instagram = Column(String(60))
    youtube = Column(String(60))
    
    # Metadata
    sincronizado = Column(Boolean, default=False)
    somente_leitura = Column(Boolean, default=False)
    deletado = Column(Boolean, default=False)
    data_criacao = Column(DateTime(timezone=True), server_default="now()")
    data_atualizacao = Column(DateTime(timezone=True), onupdate="now()")

    # Relacionamentos
    usuarios = relationship("Usuario", back_populates="cliente")

    @property
    def ativo(self) -> bool:
        return not self.deletado

Modelo Usuario (refatorado)

# app/models/central/usuario.py
from uuid import uuid4
from sqlalchemy import Column, String, Boolean, DateTime, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship

from app.db.central import CentralBase


class Usuario(CentralBase):
    __tablename__ = "usuario"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    cliente_id = Column(UUID(as_uuid=True), ForeignKey("cliente.id", ondelete="CASCADE"), nullable=False)
    
    usuario = Column(String(255), nullable=False, index=True)
    email = Column(String(255), index=True)
    senha_hash = Column(String(255), nullable=False)
    id_cadastro = Column(String(50))
    
    # Metadata
    ativo = Column(Boolean, default=True)
    deletado = Column(Boolean, default=False)
    data_criacao = Column(DateTime(timezone=True), server_default="now()")
    data_atualizacao = Column(DateTime(timezone=True), onupdate="now()")

    # Relacionamentos
    cliente = relationship("Cliente", back_populates="usuarios")

    __table_args__ = (
        {"mysql_charset": "utf8mb4"},
    )

---

Fase 4: Atualizar Repositórios

Ajustes necessários

1. Buscar por ID: usar UUID em vez de int 2. Consultas: ajustar campos renomeados 3. Inserções: incluir id=uuid4() ou default=uuid4

# Exemplo: buscar cliente por CNPJ
def get_cliente_by_cnpj(db: Session, cnpj: str) -> Cliente | None:
    return db.query(Cliente).filter(Cliente.cnpj == cnpj).first()


# Exemplo: criar usuário
def create_usuario(db: Session, dados: dict) -> Usuario:
    from uuid import uuid4
    usuario = Usuario(id=uuid4(), **dados)
    db.add(usuario)
    db.commit()
    db.refresh(usuario)
    return usuario

---

Fase 5: Migration para Tabelas Existentes

Tabela: cliente → usuario

Mudanças: 1. Renomear para usuario (ou manter cliente_usuario se necessário) 2. Remover prefixo usu_ 3. UUID como PK 4. Adicionar cliente_id (FK para cliente.id) 5. Renomear usu_senha → senha_hash 6. Adicionar ativo, data_criacao, data_atualizacao

Script SQL

-- Migration: cliente_usuario -> usuario
BEGIN;

-- 1. Criar nova tabela
CREATE TABLE public.usuario_new (
    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    cliente_id uuid REFERENCES cliente(id) ON DELETE CASCADE,
    
    usuario varchar(255) NOT NULL,
    email varchar(255),
    senha_hash varchar(255) NOT NULL,
    id_cadastro int,
    
    ativo boolean DEFAULT true,
    deletado boolean DEFAULT false,
    data_criacao timestamp with time zone DEFAULT now(),
    data_atualizacao timestamp with time zone DEFAULT now()
);

-- 2. Migrar dados (gerando hash se necessário)
INSERT INTO usuario_new (id, cliente_id, usuario, email, senha_hash, id_cadastro, data_criacao)
SELECT 
    gen_random_uuid(),
    (SELECT id FROM cliente WHERE cod_cliente = cu.cod_cliente LIMIT 1),
    cu.usu_usuario,
    cu.usu_email,
    COALESCE(cu.usu_senha, ''),  -- IMPORTANTE: deve ser hash!
    cu.id_cadastro,
    now()
FROM cliente_usuario cu;

-- 3. Dropar e renomear
DROP TABLE cliente_usuario CASCADE;
ALTER TABLE usuario_new RENAME TO usuario;

-- 4. Criar índices
CREATE INDEX idx_usuario_usuario ON usuario(usuario);
CREATE INDEX idx_usuario_email ON usuario(email);
CREATE INDEX idx_usuario_cliente ON usuario(cliente_id);

COMMIT;

---

Checklist de Segurança

• [ ] Senhas hasheadas: nunca texto puro
• [ ] CNPJ único: constraint UNIQUE
• [ ] Soft delete: campo deletado boolean
• [ ] Timestamps com timezone: timestamp with time zone
• [ ] UUID para PK: evita enumerabilidade
• [ ] Dados sensíveis: não retornados em APIs

---

Comando de Verificação

# Verificar estrutura da tabela
\d+ cliente

# Verificar índices
SELECT indexname, indexdef FROM pg_indexes WHERE tablename = 'cliente';

# Verificar constraints
SELECT conname, contype FROM pg_constraint WHERE conrelid = 'cliente'::regclass;

---

Errors Comuns

---

Quick Reference

-- Gerar UUID
gen_random_uuid()

-- Timestamp com timezone
now()::timestamp with time zone

-- Boolean
true / false (não 'S'/'N')

-- Migration básica
ALTER TABLE nome ADD COLUMN nova_coluna tipo;
ALTER TABLE nome DROP COLUMN coluna_velha;
ALTER TABLE nome RENAME COLUMN velha TO nova;

---

Próximos passos: Após criar a skill, executar migration para as tabelas cliente e cliente_usuario seguindo os padrões definidos.